PyCharter + Prefect

Prefect owns orchestration — scheduling, retries, observability, the flow/task graph. PyCharter owns the contract — the thing Prefect has no primitive for. They compose without overlap: Prefect orchestrates, PyCharter validates.

pip install pycharter[prefect]

The adapter targets Prefect 3 (prefect>=3.0,<4; verified against 3.7). A runnable end-to-end flow lives at examples/prefect_reference_flow.py.

pycharter.contrib.prefect gives you ready task factories. Each captures its contract/config once and returns a Prefect Task; **task_kwargs pass straight through to @task, so you set Prefect-native retries, retry_delay_seconds, cache_policy, and tags per task.

Both sides, one contract

The payoff is that the producer-side gate and the post-load quality check derive from the same versioned contract, so they cannot drift.

from prefect import flow
from pycharter import QualityCheckOptions
from pycharter.contrib.prefect import (
    pycharter_preflight_task,
    pycharter_quality_task,
    pycharter_validate_task,
)

CONTRACT = "contracts/orders.yaml"  # one artifact, both ends

preflight = pycharter_preflight_task(config="pipelines/orders.yaml")
validate = pycharter_validate_task(  # producer-side, per record
    contract=CONTRACT, on_invalid="filter", retries=2
)
quality = pycharter_quality_task(  # post-load, whole batch
    contract=CONTRACT,
    options=QualityCheckOptions.monitoring(),
    publish_artifact=True,  # scores/breaches in the Prefect UI
)


@flow(name="orders-etl")
def orders_etl():
    preflight()  # fail before any I/O
    raw = extract_from_source()  # your Prefect task
    clean = validate(raw)  # gate records in-flight
    load_to_warehouse(clean)  # your Prefect task
    snapshot = read_warehouse_snapshot()  # your Prefect task
    quality(snapshot)  # post-load DQ; fails flow on breach

The validate task

pycharter_validate_task(contract=…, on_invalid=…) takes a list[dict] and returns a list[dict]. on_invalid chooses the failure policy:

on_invalid	Behaviour
"raise" (default)	Raise ContractValidationError if any record fails; otherwise return the coerced valid records.
"filter"	Drop failing records; return the coerced valid ones.
"tag"	Return every record in input order; failing records gain a _violations key.

The Validator is built once at factory time and reused per call (thread-safe), so a high-throughput flow pays the compile cost only once.

Sync vs. async — the one gotcha

Validator.validate / validate_records are pure-sync and safe in any task. The pipeline is different: pycharter.run() calls asyncio.run() internally and crashes inside a running event loop. So:

• In an async flow, pycharter_run_task(...) (default is_async=True) awaits Pipeline.run() directly — loop-safe.
• In a sync flow, pass is_async=False to use pycharter.run().
• The quality task offloads the synchronous check_quality to a worker thread (anyio.to_thread), so it never blocks the flow's loop.

Scope: keep Prefect as the orchestrator

Use PyCharter for validation and quality inside your Prefect tasks. Reach for pycharter_run_task only when you deliberately want to run a whole PyCharter pipeline as a single task — don't rebuild Prefect's scheduling, retries, or connection management with PyCharter's own extractors/loaders.

See PyCharter vs. the alternatives for where PyCharter fits next to Airflow, Dagster, and warehouse-side quality tools.