Contract → ETL → quality¶

PyCharter separates what good data means (the data contract) from where you enforce it (pipeline steps, APIs, streams, and scheduled jobs). The contract is the single specification; the pipeline does not fork rules—it opts in to parts of the contract at each step.

This page is the runtime workflow (what runs, in what order). For platform lifecycle (author → parse → store → deploy), see The complete data journey. For “contract quality vs pipeline quality” APIs, see Contract vs pipeline quality.

At a glance¶

flowchart TB
  subgraph spec [Single source of truth]
    C[("Data contract YAML")]
    C --> JS[json_schema]
    C --> CR[coercion_rules]
    C --> VR[validation_rules]
    C --> FB[field_bindings]
    C --> DQ[data_quality]
    C --> MD[metadata / SLA]
  end

  subgraph enforce [Where enforcement happens]
    P[ETL pipeline steps]
    API[REST / FastAPI gate]
    STR[Streaming worker]
    JOB[QualityCheck / worker jobs]
  end

  spec --> P
  spec --> API
  spec --> STR
  spec --> JOB

  P --> OUT[(Valid data / metrics / DLQ)]
  API --> OUT
  STR --> OUT
  JOB --> OUT

Contract layers (what each section does)¶

Think of the contract as stacked concerns. Only some layers are driven by pipeline contract: flags.

block-beta
  columns 3

  block:layer:3
    columns 1
    L4["data_quality.rules + targets"]
    L3["validation_rules + json_schema"]
    L2["coercion_rules"]
    L1["metadata · governance · SLA"]
  end

  block:flags:3
    columns 1
    F4["quality: true → batch at step"]
    F3["validate: true → per row"]
    F2["coerce: true → per row"]
    F1["annotate: true → per row"]
  end

Contract section	Granularity	Pipeline flag	Question it answers
`coercion_rules`	Per field / record	`coerce: true`	Can we normalize values before judging them?
`json_schema`	Per record	`validate: true`	Does the record have the right shape and types?
`validation_rules`	Per record	`validate: true`	Do business rules pass (length, enums, regex)?
`field_bindings`	Per record metadata	`annotate: true`	What concept does each field mean?
`data_quality.rules`	Batch at step	`quality: true`	Does this batch pass aggregate checks (null rate, row count)?
`data_quality.targets`	Dataset / run	`QualityCheck`, monitoring	Did we meet score / violation-rate goals?
`sla_*`	Operational	Dashboards / process	Latency, retention commitments (metadata)

Reference layout: template_contract.yaml in the package, or Templates overview.

Pipeline topology¶

A typical batch pipeline attaches the same or different contracts at stage boundaries. Transforms sit between gates.

flowchart LR
  subgraph sources [Sources]
    HTTP[HTTP API]
    FILE[Files / DB]
    MSG[Kafka / queues]
  end

  subgraph pipeline [Pipeline]
    E[Extract step]
    T[Transform]
    L[Load step]
  end

  subgraph contracts [Contract refs]
    CS[Source contract]
    CT[Target contract]
  end

  sources --> E
  E -->|"contract: validate, coerce?"| T
  T --> L
  L -->|"contract: validate, coerce, quality?"| WH[(Warehouse / file)]

  CS -.-> E
  CT -.-> L

Common pattern

Step	Typical flags	Why
Extract	`validate: true`, `coerce: false`, `on_error: quarantine`	Reject bad source rows early
Transform	(no contract) or light contract	Shape mapping only
Load	`validate: true`, `coerce: true`, `quality: true`	Normalize to target + batch DQ before persist

Quarantine requires an explicit contract.dlq.target on the step. A policy without a destination is rejected, including in existing manifests and previews. See the quarantine policy and migration.

Step `contract:` block¶

Each extract, transform, join, union, or load step may include:

contract:
  ref:
    type: file          # or store: name + version
    path: ./contracts/orders.yaml
  coerce: true
  validate: true
  annotate: false
  quality: false        # opt in to data_quality.rules on this batch
  on_error: fail        # fail | warn | skip | quarantine

flowchart TD
  REF[Resolve ref → load contract]
  REF --> BATCH[Incoming record batch]

  BATCH --> LOOP{For each record}

  LOOP --> C{coerce?}
  C -->|yes| CO[Apply coercion_rules]
  C -->|no| V0[Record unchanged]
  CO --> V0

  V0 --> V{validate?}
  V -->|no| A0[Keep record]
  V -->|yes| VAL[Validator: schema + validation_rules]
  VAL -->|valid| A0
  VAL -->|invalid| OE{on_error}
  OE -->|fail| STOP[[Step fails]]
  OE -->|skip / quarantine| DROP[Drop record]
  OE -->|warn| A0

  A0 --> A{annotate?}
  A -->|yes| AN[_concept_annotations]
  A -->|no| EMIT[Emit record]
  AN --> EMIT

  EMIT --> LOOP
  LOOP -->|batch done| Q{quality?}
  Q -->|no| DONE[Step complete]
  Q -->|yes| DQR[Evaluate data_quality.rules on batch]
  DQR -->|severity error| QFAIL[[StepQualityError]]
  DQR -->|pass / warnings| DONE

Validation vs inline quality

Invalid rows (schema / validation_rules) are handled by on_error (drop, quarantine, or fail the step). data_quality rule failures with severity: error fail the step but do not route rows to a DLQ—that separation is intentional (validation owns record flow; quality owns batch assertions).

Implementation: apply_step_contract and apply_step_contract_block in the pipeline generator and semantic runtime adapters.

Three ways to run quality¶

PyCharter exposes three complementary quality paths. Use one or combine them.

flowchart TD
  START([You have data + a contract])

  START --> Q1{Need per-row gate<br/>in ETL?}
  Q1 -->|yes| S1[Step contract.validate + coerce]
  Q1 -->|no| Q2

  S1 --> Q2{Need batch rules from<br/>contract data_quality?}
  Q2 -->|yes| S2[Step contract.quality: true]
  Q2 -->|no| Q3

  S2 --> Q3{Need run assurance<br/>after load?}
  Q3 -->|yes| S3[load quality_checks YAML]
  Q3 -->|no| Q4

  S3 --> Q4{Need score / violations<br/>over time?}
  Q4 -->|yes| S4[QualityCheck / POST /quality/check / worker]
  Q4 -->|no| END([Done])

  S4 --> END

Path	Granularity	Configuration	Run artifact
Step validate	Row	`contract.validate`	Pass/fail per record; drops or step error
Step quality	Batch at step	`contract.quality` + `data_quality.rules`	`StepQualityReport`; error fails step
Load quality_checks	Run / dataset	`quality_checks` on load config	`pipeline_quality_report`
QualityCheck	Full dataset	Python / REST / worker job	Score 0–100, violations, thresholds

Details and REST field names: Contract vs pipeline quality, Pipeline quality, Quality monitoring tutorial.

End-to-end story (batch ETL)¶

sequenceDiagram
  participant Author
  participant Store as Contract store
  participant Pipe as Pipeline runner
  participant WH as Target

  Author->>Author: Write orders_contract.yaml
  Author->>Store: Register name + version
  Pipe->>Store: Resolve contract at extract
  Pipe->>Pipe: Extract batch
  Pipe->>Pipe: coerce / validate (source contract)
  Pipe->>Pipe: Transform rename / convert
  Pipe->>Store: Resolve target contract at load
  Pipe->>Pipe: coerce / validate / quality (target)
  Pipe->>WH: Load surviving rows
  Note over Pipe: Optional load quality_checks
  Author->>Pipe: QualityCheck job (scheduled)
  Pipe-->>Author: Score + violations

Author the contract (or start from Templates / pycharter contract sketch).
Register in the contract store or keep files beside the pipeline.
Extract with validate: true on the source contract.
Transform (optional contract on intermediate shape).
Load with coerce: true, validate: true, and optionally quality: true.
Optionally add load quality_checks for operational row-count checks.
Optionally schedule QualityCheck.run() or a validation worker for dashboards and data_quality.targets — see Periodic quality monitoring.

Tutorial walkthrough: Building ETL pipelines.

Streaming (same contract block)¶

The streaming worker reuses the same contract: shape on each stream in streaming.streams[] (per-message coerce/validate, DLQ, modes).

flowchart LR
  KIN[Kafka topic in]
  SW[stream worker]
  C[contract block]
  KOUT[Validated topic]
  DLQ[DLQ topic]

  KIN --> SW
  C -.-> SW
  SW --> KOUT
  SW --> DLQ

Example: examples/streaming/orders/stream-worker.yaml
Reference: Streaming worker config
Tutorial: Streaming contract validation

There is no pycharter stream init template pack yet; copy the example or adapt batch extract YAML — see Templates — streaming.

APIs and services (same contract, different entry)¶

mindmap
  root((Data contract))
    ETL
      pipeline init
      step contract block
      load quality_checks
    Streaming
      pycharter stream run
      per message validate
    Ad hoc
      Validator
      QualityCheck
      POST quality/check
    Platform
      contract store
      REST ops/templates
      Web UI

Entry	Docs
Contract-first authoring	Contract-first workflow
YAML starters	Templates overview
Platform lifecycle	Complete data journey
Quality types comparison	Contract vs pipeline quality

What PyCharter does not schedule for you¶

data_quality.monitoring_enabled, per-rule schedule, and SLA fields are metadata and design intent. Execution still requires your scheduler (cron, Airflow, Dagster) or the validation worker — see Periodic quality monitoring.

Next steps¶

Contract-first workflow — define and store
Templates overview — contract and pipeline YAML starters
Building ETL pipelines — hands-on pipeline
Contract vs pipeline quality — APIs and run metrics