Streaming Contract Validation¶

The streaming validation engine validates live records against data contracts, routes invalid records to a DLQ, and writes valid records to your chosen sink — in five lines of Python or via a long-lived stream worker process.

Kafka is optional. The engine is source/sink agnostic: bundled examples include WebSocket→file (streaming-massive-file), WebSocket→Kafka (streaming-massive-kafka), Kafka→file (streaming-kafka-file), and SQS→Kafka (streaming-ecommerce-sqs). See src/pycharter/data/seed/examples/README-streaming.md. Use kafka when you need cross-language fan-out; use file or postgres/database sinks for local dev and direct persistence.

This guide covers the production surface. For a five-minute walkthrough see the Streaming Validation Tutorial. For the implementation plan and design decisions see Streaming Validation Plan.

Experimental → stable: Run tab + dev stream-worker supervisor, contract.annotate, database sinks, durable window checkpoints, and quality reducers are implemented. Remaining production deployment guidance (k8s pycharter stream run Deployments, exactly-once constraints) stays documented below.

When to use it¶

You want…	Use this.
Validate Kafka messages in your Python service before processing them	Library mode (`pycharter.kafka.consume`)
Validate a topic across many consumers with different languages	Worker mode (`pycharter stream run`) — the worker republishes valid records to a separate topic
Roll out validation gradually without breaking downstream consumers	Operating modes — start in `detect`, move to `shadow`, finalize at `enforce`
Replay the DLQ after fixing a contract	Replay CLI (`pycharter stream replay`) — re-validates DLQ envelopes against a chosen contract version, with `--dry-run` for previewing
Edit a stored worker config in the browser	Editing stored configs in the web UI — the unified pipeline editor opens streaming configs

Library mode¶

Three public entry points, all in pycharter.kafka:

from pycharter import kafka

# 1. Producer-side helper — stamps the contract headers automatically.
await kafka.produce(
    topic="orders",
    value={"order_id": "abc", "amount": 50.0, "ok": True},
    contract="orders@2",
    bootstrap_servers="kafka:9092",
    key="user-7",  # preserved for partition routing
)

# 2. Consumer-side iterator — only validated records reach the loop.
async for record in kafka.consume(
    topic="orders",
    bootstrap_servers="kafka:9092",
    group_id="my-service",
    contract_dir="contracts/orders",  # or contract_store=..., or validator_factory=...
):
    process(record.payload)

# 3. Sync wrapper — for notebooks / legacy ETL that can't go async.
for record in kafka.consume_sync(
    topic="orders",
    bootstrap_servers="kafka:9092",
    group_id="notebook",
    contract_dir="contracts/orders",
):
    process(record.payload)

Validator factories¶

consume() accepts exactly one of three validator-factory inputs:

contract_dir=Path(...) — single directory holding one contract; convenience for one-topic-one-contract setups.
contract_store=ContractStoreClient(...) — a real registry. The engine resolves each (contract_id, version) from the message headers and caches the resulting Validator in-process.
validator_factory=callable — direct callable for advanced use (custom registries, test fakes).

Static fallback for migration¶

If some producers can't stamp the headers yet:

async for record in kafka.consume(
    topic="orders",
    contract_dir="contracts/orders",
    static_contract="orders@2",                 # ContractRef or "name@version"
    on_missing_header="fallback_to_static",     # opt-in policy
    ...
):

Records with the headers are validated against whichever version they declare; records without the headers fall back to orders@2.

Worker mode¶

pycharter stream run config.yaml

The worker is the same engine driven by config:

id: example/orders
version: "1.0.0"
streams:
  - name: orders-validator
    source:
      type: kafka
      topic: orders
      bootstrap_servers: kafka:9092
      consumer_group: orders-validator
    contract:
      ref: { type: file, path: /app/contracts/orders }
      resolution: static
    load:
      target: { type: kafka, topic: orders-validated, bootstrap_servers: kafka:9092 }
    dlq:
      target: { type: kafka, topic: orders-dlq, bootstrap_servers: kafka:9092 }
    mode: enforce
    delivery: at_least_once
observability:
  health_port: 8080

One process can run many streams. SIGTERM triggers graceful shutdown — every stream finishes its in-flight message, sinks flush, the process exits 0.

stream run accepts the same variable overrides as pipeline run: --var KEY=VALUE (repeatable) merges over the config's variables: block before ${VAR} resolution.

Validate before you run¶

pycharter stream validate config.yaml            # local file
pycharter stream validate --name orders/prod     # stored config
pycharter stream validate config.yaml --var REGION=eu-west-1

stream validate is the streaming sibling of pipeline validate: it parses the config, resolves every stream's contract (file directories are loaded and parsed; contract stores are connected and static refs resolved), checks source/sink types and their required keys, parses window specs, and applies the exactly_once constraints — all without any Kafka I/O. Issues are collected per stream:

Validation FAILED: 2 preflight issue(s)
  [orders-validator] contract: contract directory not found: /app/contracts/orders
  [trades] load: load.target type 'file' requires 'path'

Exit codes: 0 clean, 1 preflight issues, 2 the config failed to load or parse. The same checks are available in Python as StreamWorker.from_file(...).preflight().

Preview with --dry-run¶

pycharter stream run config.yaml --dry-run --max-records 100

--dry-run consumes, resolves, and validates without writing to any sink — no valid-topic produce, no DLQ writes, no extraction side effects, and window checkpoints are neither restored nor overwritten. Kafka sources are switched to an ephemeral consumer group (pycharter-preview-<random>), so a preview never advances a real group's offsets. --max-records N bounds the preview; per-stream totals (consumed / valid / invalid) are logged at exit. Dry-run is rejected for exactly_once streams — a transactional preview would still commit offsets.

Discover and inspect configs¶

pycharter stream list                    # scan for worker.yaml files
pycharter stream list --store            # list stored configs
pycharter stream inspect config.yaml     # summary: streams, contracts, sinks
pycharter stream inspect --name orders/prod

Start/stop from the editor (dev)¶

When PYCHARTER_STREAM_SUPERVISOR_ENABLED=1 (default when ENVIRONMENT is not production), the API can spawn a local stream worker subprocess for a saved config. In production (or when PYCHARTER_STREAM_JOBS_ENABLED=1 with the supervisor off), Start enqueues a streaming_jobs row instead — a separate pycharter stream launch process claims jobs and spawns pycharter stream run children.

Path	When	Command
Dev supervisor	`ENVIRONMENT != production` (default)	API spawns subprocess
Job queue	Production / `PYCHARTER_STREAM_JOBS_ENABLED=1`	`pycharter stream launch`

API endpoints:

POST /api/v1/streaming/configs/{name}/{version}/start — supervisor spawn or enqueue (202)
POST /api/v1/streaming/configs/{name}/{version}/stop — cancel job + SIGTERM
POST /api/v1/streaming/jobs — explicit enqueue
GET /api/v1/streaming/jobs/stats — queue depth for the Run tab / ops
POST /api/v1/streaming/jobs/{job_id}/cancel

Launcher (production):

pycharter stream launch --db-url $PYCHARTER_DATABASE_URL

Run one or more launcher replicas (k8s Deployment). This is not pycharter worker — the job worker handles finite validation/pipeline jobs; the stream launcher handles long-lived stream workers.

Explicitly invoked owned SQL queue methods provide local PID/cancellation state guarded by the exact current fence. The current launcher does not yet carry those fences; a queue acknowledgement or stored PID does not establish that an external process stopped or authorize a later process signal.

Opt-in seed hook: set post_seed.run_streaming: true in an example manifest.yaml to start (or print) the stream worker after pycharter db seed.

See the config schema reference for every field.

Editing stored configs in the web UI¶

Worker configs can be persisted in the streaming-config registry (/api/v1/streaming/configs, keyed by stream_name + stream_version) and managed from the web UI. The configs list lives at Pipelines → Streaming → Configs (/pipelines/streaming/configs). New config opens a dialog that collects the config's name, version, and a stream template, stores the seeded config, and continues into the unified pipeline editor — the same surface used for batch pipeline configs and, since 2026-06-06, the single place where stored streaming configs are edited (the former standalone raw-YAML Edit page is retired).

Opening a config in the editor¶

Three paths lead to the same editor tab:

Editor Browse tab. In the editor's pipeline mode, the Browse tab shows a collapsible Streaming section below the batch pipeline tree. Each row carries a streaming kind badge. A single click opens the config in a preview tab; a double click opens a permanent tab.
Configs list. Each row on /pipelines/streaming/configs has an Open in editor action.
Deep link. The editor route is the batch route plus a query parameter: /editor/pipelines/<name>/<version>?kind=streaming. Without the parameter, the same URL opens the batch pipeline of that name and version. The parameter survives both in-app navigation and pasting the URL into a fresh browser session.

Editing on the canvas¶

Each entry in streams[] renders as a three-node chain — source → process → sink — plus a DLQ sidecar attached to the process node. Streams are independent horizontal lanes; no edges cross between them. All four nodes are editable:

Inspectors. Selecting a node opens its inspector. Each inspector has a Form tab for the modeled fields and a Raw JSON tab (parse-guarded Apply) for everything else:
- Source — source type (Kafka, RabbitMQ, HTTP, WebSocket, SSE, SQS) plus type-specific fields.
- Process — stream name, contract reference, validate / coerce flags, mode, delivery, on_error. Its Raw JSON tab shows the full stream body, so unmodeled per-stream keys are edited here.
- Sink and DLQ — the load / dlq target block (Kafka topic or JSONL file).

Fields the forms do not model stay editable in the Raw JSON tabs, and unknown keys survive every save verbatim. - Palette templates. The Components tab offers three drag-and-drop stream templates: Kafka → Kafka, Kafka → File, and Empty stream. Dropping one onto the canvas appends a complete, schema-valid chain (every template includes a DLQ block because the schema requires one). Templates reference a placeholder contract (my_contract@1.0.0) — it passes save-time validation, but you must point it at a real contract in the process inspector before a worker can run the stream. - Deleting. Deleting the process node removes the whole stream (its chain and DLQ sidecar). Deleting the DLQ sidecar drops only the dlq block. Source and sink nodes cannot be deleted on their own — the chain shape is structural. To re-add a deleted DLQ, add the dlq: block back in the Form tab YAML (see below). - Node positions. Layout adjustments persist locally in the browser (per config name + version), not in the stored config — moving nodes never dirties the workspace, and positions do not follow the config to other machines.

Editing as YAML (Form tab)¶

The Phase-1 workflow remains fully supported. The Form tab shows the whole config as YAML (or JSON via the format toggle):

Apply parses the text, merges it into the in-memory config, re-projects the canvas, and marks the workspace dirty. Syntax errors and an empty streams array are rejected inline without touching the loaded config. Apply never contacts the server.
Save (toolbar button) persists the applied config with an update to the loaded (stream_name, stream_version). There is no save modal — the identity is locked when the config is loaded, so editing stream_name or stream_version in the YAML does not rename or fork the stored config. To create a new config, use the New config dialog on the configs list page (name + version + template), which stores a seeded config and opens it in the editor.

Sections of the config the canvas does not model (unknown stream keys, metadata) ride along verbatim: loading a config and saving it without edits produces an equivalent stored body. metadata edits made in the YAML persist on save (see the limitations note below for the one caveat).

Bottom panel — Observability and Source Tap¶

Streaming editor tabs replace the batch bottom panel (Variables / Settings / Run) with two streaming-specific tabs:

Observability — a structured editor for the worker-level observability block (health/metrics ports, log format) shared by every stream in the config. Edits mark the workspace dirty and persist through the normal Save.
Source Tap — a bounded, read-only dry-run of one selected stream's source against its contract (the same engine as the standalone /pipelines/streaming/configs/tap page). The tap runs against the last saved version of the config — a hint appears when the workspace has unsaved edits, since they are not reflected until you Save.

Validation on save¶

Every save goes through the server-side validation guard: the API validates the assembled config with StreamingConfig.model_validate and rejects bad shapes with 422 before anything is persisted. A rejected save shows a Save failed: … message inline under the toolbar, and the workspace stays dirty — fix the config, Apply (if editing YAML), and Save again without losing edits.

Known limitations¶

Limitation	Detail
`metadata` cannot be cleared	The block persists on save (added 2026-06-06), but an update cannot remove it entirely — an absent or null `metadata` on update means "leave the stored value unchanged". To blank it out, save an explicit empty-valued block.
Node positions are per-browser	Canvas layout is stored in browser local storage, not in the config — other machines and other users see the default lane layout.

Operating modes¶

Three modes ship out of the box. Configure on the stream (mode: field) or via the mode= kwarg on consume().

Mode	What happens to invalid records
`enforce` (default)	Routed to DLQ. The application loop never sees them.
`detect`	Passed through to the application loop and counted in metrics. No DLQ writes. Use this for safe rollouts — observe what would break before flipping to `enforce`.
`shadow`	Passed through to the application loop and routed to the DLQ. Use this when downstream still needs the bad record while you investigate.

Recommended rollout: ship in detect, watch the violation rate in pycharter_streaming_messages_total{outcome="validation_failed"}, flip to shadow for a release, then finalize in enforce.

Delivery semantics¶

Mode	Behavior on crash
`at_least_once` (default)	Commit offset after the sink confirms. Crash before commit → message reprocessed → duplicate in sink. Validation is deterministic so duplicates are byte-identical.
`at_most_once`	Commit offset immediately on consume, before validation. Crash during validation → message lost.

For data validation at-least-once is the only sane default — losing a message means a bad record might escape unnoticed, defeating the purpose. Switch to at_most_once only when downstream demands no-duplicates and you accept the loss tradeoff.

DLQ envelope¶

Every DLQ record carries a structured envelope so replay tooling has full context:

{
  "original_payload": {"order_id": "abc", "amount": -1, "ok": false},
  "original_headers": {
    "pycharter-contract": "orders",
    "pycharter-contract-version": "2",
    "x-trace": "..."
  },
  "error_reason": "validation_failed",
  "error_message": "record failed contract validation",
  "validation_errors": [
    {"path": "$.amount", "message": "must be >= 0"}
  ],
  "contract_id": "orders",
  "contract_version": "2",
  "source": {
    "topic": "orders",
    "partition": 3,
    "offset": 12345,
    "timestamp_ms": 1746000000000
  },
  "dlq_timestamp": "2026-05-09T01:00:00+00:00"
}

Headers on the DLQ message itself include pycharter-dlq-reason: validation_failed (or decode_error, no_contract_header, etc.) so DLQ consumers can filter without parsing the JSON body.

Observability¶

The worker (and library mode if you wire the observers manually) exposes:

Endpoint	Purpose
`GET /healthz`	Liveness probe — 200 while the listener is up.
`GET /readyz`	Readiness probe — 200 only when every stream's registry breaker is closed. 503 + JSON body of unready streams otherwise.
`GET /metrics`	Prometheus exposition.

Prometheus metrics:

Metric	Type	Labels
`pycharter_streaming_messages_total`	counter	`stream`, `outcome`
`pycharter_streaming_validate_latency_seconds`	histogram	`stream`
`pycharter_streaming_circuit_breaker_state`	gauge	`resource`

Outcomes (counter label values): valid, decode_error, no_contract_header, unknown_contract, unknown_contract_version, validation_failed, oversized_payload, registry_unavailable.

The WindowedQualityReducer (also wired by default in worker mode) maintains rolling counters per stream over configurable windows (default 1m / 5m / 1h). Snapshot via reducer.snapshot() returns one WindowSnapshot per (stream, window) with valid/invalid counts, violation rate, and p99 validate latency.

Failure handling¶

Failure	Where detected	Behavior
Decode error (malformed JSON)	source	Route to DLQ as `decode_error`; never crash.
Header missing, `strict` policy	resolver	Route to DLQ as `no_contract_header`.
Unknown contract id / version	resolver	Route to DLQ as `unknown_contract`.
Validation fails	validator	Route to DLQ as `validation_failed` with errors.
Sink write fails (transient)	sink	Internal retry; do not commit offset.
Registry unreachable	resolver	Block on retry; trip circuit breaker after timeout (default 60s) → `/readyz` returns 503; k8s removes pod from service.
`SIGTERM`	runtime	Drain in-flight message, flush sinks, commit offsets, exit 0.

The engine never silently drops a record: every failure is either DLQed or surfaces as a worker unready state.

Authentication¶

Kafka auth is configured on every source/sink config or via consume() / produce() kwargs:

Mode	Required fields
PLAINTEXT (dev)	(none — defaults)
SASL/SCRAM-SHA-256 / 512	`security_protocol: SASL_PLAINTEXT` (or `SASL_SSL`), `sasl_mechanism`, `sasl_username`, `sasl_password`
mTLS	`security_protocol: SSL`, `ssl_cafile`

Never inline credentials in YAML. The recommended pattern is to load them from the environment:

sasl_password: ${KAFKA_PASSWORD}    # parsed by your env interpolation step

…or pass them in via os.environ to consume()/produce() directly.

Migration from the legacy `KafkaExtractor`¶

pycharter.pipeline_generator.extractors.KafkaExtractor continues to exist and the legacy batch interface (extract_streaming) is unchanged. The streaming validation engine adds a new per-record interface (iter_messages + commit + close) that every extractor inherits from BaseExtractor. New code should prefer the streaming surface; legacy pipelines keep working with no edits.

If you previously rolled your own validation around KafkaExtractor:

# Before
extractor = KafkaExtractor(topics=["orders"], ...)
async for batch in extractor.extract({}):
    for record in batch:
        result = my_validator.validate(record)
        if result.is_valid:
            process(record)
        else:
            send_to_dlq(record, result.errors)
    await extractor.acknowledge(batch_index, success=True)

…replace it with:

# After
async for record in pycharter.kafka.consume(
    topic="orders",
    bootstrap_servers="kafka:9092",
    group_id="my-service",
    contract_dir="contracts/orders",
):
    process(record.payload)

The engine handles the validation, the DLQ routing, the offset commit, and the failure-mode handling.

Gateway mode (Pattern A)¶

The same engine can run as a standalone gateway that sits between producers and consumers, republishing valid records to a separate topic. This is what "Pattern A" referred to in the design discussion.

The gateway is just the worker (pycharter stream run) with load.target pointed at a Kafka topic — there's no separate code path. Topology:

producer ──► orders-raw ──► [pycharter worker] ──► orders-validated ──► consumer-1
                                   │                                    consumer-2 (any language)
                                   └──► orders-dlq                      consumer-3

Worker config:

streaming:
  streams:
    - name: orders-gateway
      source:
        type: kafka
        topic: orders-raw
        bootstrap_servers: kafka:9092
        consumer_group: pycharter-gateway
      contract:
        ref: { type: file, path: /app/contracts/orders }
        resolution: static
      load:
        target: { type: kafka, topic: orders-validated, bootstrap_servers: kafka:9092 }  # downstream consumers read here
      dlq:
        target: { type: kafka, topic: orders-dlq, bootstrap_servers: kafka:9092 }
      mode: enforce
      delivery: at_least_once

When to choose gateway over library mode:

	Library	Gateway
Languages	Python only (consumer imports `pycharter.kafka.consume`)	Any (downstream reads plain Kafka)
Topology	One topic, in-process validation	Two topics + a worker process
Latency	One hop fewer	One additional hop (~10–50 ms)
Operational footprint	None — runs inside your service	One pod / deployment to manage
Observability scope	Per-consumer	Centralised at the gateway

Most teams pick library mode first and only adopt the gateway when they introduce non-Python consumers downstream.

Schema compatibility check¶

When a contract version changes, you can preview the impact before producers start stamping the new version. The schema-compatibility checker diffs two JSON schemas and classifies the change:

from pycharter.streaming._compatibility import (
    Compatibility,
    check_compatibility,
)

old_schema = {
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "amount": {"type": "number"},
    },
    "required": ["order_id", "amount"],
}
new_schema = {
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "amount": {"type": "number"},
        "currency": {"type": "string"},  # added optional field
    },
    "required": ["order_id", "amount"],
}

report = check_compatibility(old_schema, new_schema)
print(report.verdict)  # Compatibility.FULL — adding optional field is safe
for change in report.changes:
    print(change.path, change.kind, change.detail)

The verdict is one of:

Verdict	Meaning
`FULL`	Both directions safe. Producers can flip immediately.
`BACKWARD`	New consumers can read old data. Roll out new consumers first, then producers.
`FORWARD`	Old consumers can still read new data. Producers can flip whenever.
`NONE`	Breaking change. Bump the major version; rollout requires consumer + producer in lockstep.

The check is conservative — it errs on the side of flagging changes as incompatible. Treat the report as a starting point for review, not the final word. Limitations: no $ref resolution, no oneOf / anyOf / allOf composition. Treat reports involving those features as "investigate manually".

Replay CLI¶

When a contract version changes, the records that ended up in the DLQ under the old contract may now be valid. pycharter stream replay reads from the DLQ topic, parses each envelope, and re-validates the original payload against the contract you specify:

# Preview: count how many DLQ records would now pass validation under
# orders@3, but don't write anything.
pycharter stream replay \
  --mode dlq \
  --source-topic orders-dlq \
  --bootstrap-servers kafka:9092 \
  --contract-file contracts/orders-v3.yaml \
  --contract orders@3 \
  --dry-run

Replay complete: scanned=120 valid=85 invalid=35 skipped=0 withheld=0 (dry_run=True)

Drop --dry-run and add --target-topic orders-validated to forward the now-valid records into the validated topic. Replay is non-destructive — it never deletes from the DLQ; rerunning is safe.

--contract-file takes a single kind: DataContract manifest and is the form that carries field_bindings, so replayed records go through the same privacy gate as a live stream: --privacy-mode detect (the default) reports flagged fields, --privacy-mode enforce tokenizes pii / redacts restricted (add --clearance for a sink allowed the clear value; the org secret comes from PYCHARTER_KB_ORG_SECRET). Records that cannot be handled safely under enforce are withheld — counted in withheld=, never forwarded in clear. The deprecated split-directory --contract-dir form carries no bindings: it still validates, but enforce refuses it and detect warns that the privacy report will be empty. See Privacy flags.

--mode range reads raw messages from a source topic instead of parsing DLQ envelopes (useful for backfilling new contract versions across historical data).

Stability¶

pycharter.kafka and pycharter.streaming are experimental through the 0.x series. The public API is unlikely to change but minor breaking adjustments may land between minor releases. Both modules graduate to stable at 1.0 once two consecutive patch releases ship without breaking changes.

If you need stability today, pin to an exact pycharter version. Breaking changes will be called out in the release notes.