Streaming Contract Validation

Validate a Kafka topic against a pycharter contract in five minutes.

PyCharter ships a streaming validation engine that consumes any topic, validates each message against a contract, routes the bad ones to a DLQ, and yields the good ones to your application. Producers and consumers both stay in plain Python; no JVM, no separate worker required.

This tutorial walks through:

1. The contract.
2. The producer side (stamp the contract headers automatically).
3. The consumer side (async for record in pycharter.kafka.consume(...)).
4. The worker mode (pycharter stream run config.yaml) for multi-stream deployments.

For the full template catalog (contracts, batch pipelines, streaming worker shape), see Templates overview. A runnable worker example lives at examples/streaming/orders/stream-worker.yaml in the repository.

Prerequisites

pip install "pycharter[streaming]"

A running Kafka broker. For local development:

docker run --rm -p 9092:9092 apache/kafka:latest

1. Define the contract

Save this as contracts/orders/contract.yaml:

contract:
  name: orders
  version: "2"
json_schema:
  type: object
  required: [order_id, amount]
  properties:
    order_id:
      type: string
    amount:
      type: number
      minimum: 0

2. Produce records with the contract headers stamped

import asyncio
from pycharter import kafka


async def main() -> None:
    await kafka.produce(
        topic="orders",
        value={"order_id": "abc", "amount": 50.0},
        contract="orders@2",
        bootstrap_servers="localhost:9092",
        key="user-7",
    )


asyncio.run(main())

The produce() helper stamps two headers on every message:

Header	Value
pycharter-contract	orders
pycharter-contract-version	2

Consumers read those headers to look up the right contract version.

3. Consume with validation

import asyncio
from pycharter import kafka


async def main() -> None:
    async for record in kafka.consume(
        topic="orders",
        bootstrap_servers="localhost:9092",
        group_id="my-service",
        contract_dir="contracts/orders",
    ):
        # Only valid records reach this loop. Invalid ones flow to
        # ``orders-dlq`` automatically.
        print(f"got {record.payload}")


asyncio.run(main())

What's happening:

• The engine reads each Kafka message, decodes it, and resolves its contract from the headers.
• Valid records are yielded to your loop.
• Invalid records (failed validation, missing headers, decode errors) are routed to a DLQ topic — by default f"{topic}-dlq" — with a structured envelope (original payload, validation errors, contract reference, source position).
• Offsets are committed only after your loop has received the record (at-least-once delivery).

4. Operating modes

async for record in kafka.consume(
    topic="orders",
    mode="detect",  # or "enforce" (default) or "shadow"
    ...
):

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.

5. Static fallback for migration

If some producers cannot stamp the headers yet, configure a fallback:

async for record in kafka.consume(
    topic="orders",
    contract_dir="contracts/orders",
    static_contract="orders@2",
    on_missing_header="fallback_to_static",
    ...
):

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

6. Worker mode for multi-stream deployments

When you have several topics to validate from one process, use the worker:

# stream.yaml
streaming:
  streams:
    - name: orders-validator
      source:
        type: kafka
        topic: orders
        bootstrap_servers: kafka:9092
        consumer_group: orders-validator
      contract:
        ref: { type: file, path: 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

    - name: payments-validator
      source:
        type: kafka
        topic: payments
        bootstrap_servers: kafka:9092
        consumer_group: payments-validator
      contract:
        ref: { type: file, path: contracts/payments }
        resolution: static
      load:
        target: { type: kafka, topic: payments-validated, bootstrap_servers: kafka:9092 }
      dlq:
        target: { type: kafka, topic: payments-dlq, bootstrap_servers: kafka:9092 }

pycharter stream run stream.yaml

Each stream runs in its own asyncio task. SIGTERM triggers a graceful shutdown — every stream finishes its in-flight message, sinks flush, then the process exits.

7. Sync wrapper for notebooks and legacy ETL

If your code is sync, the same engine works through a thread-backed wrapper:

from pycharter import kafka

for record in kafka.consume_sync(
    topic="orders",
    bootstrap_servers="localhost:9092",
    group_id="my-service",
    contract_dir="contracts/orders",
):
    print(record.payload)

The wrapper runs consume() on a private thread + event loop and relays records into the calling thread. Closing the iterator (or breaking out of the loop) tears the background loop down cleanly.

What's next

• DLQ replay: read from orders-dlq to triage and reprocess. The §3.7 envelope schema in docs/guides/streaming-validation-plan.md documents every field.
• Schema evolution: producers stamp the version they wrote against; consumers validate per-message. Old and new versions can coexist on the same topic.
• Observability: streaming quality metrics, Prometheus exposition, and /healthz / /readyz HTTP endpoints land in Phase 4 of the streaming plan.
• More sources: Kafka is the first; Redis Streams, NATS, MQTT, and Pub/Sub follow the same BaseExtractor pattern.

See Streaming Validation Plan for the full architecture.