Consume a Kafka topic with a contract¶

Validate every message off a Kafka topic, send valid records to your sink, and dead-letter the rest with a structured reason. Requires pip install pycharter[streaming].

import asyncio

from pycharter import Validator
from pycharter.streaming import (
    ContractStreamProcessor,
    HeaderContractResolver,
    IteratorSink,
    MemoryDLQSink,
)
from pycharter.streaming.kafka import KafkaSource


async def main() -> None:
    # 1. Where messages come from.
    source = KafkaSource(
        bootstrap_servers="kafka:9092",
        topic="orders.v1",
        group_id="orders-validator",
    )

    # 2. How to find the right contract per message. HeaderContractResolver
    #    reads the contract id and version from message headers; pair with
    #    StaticContractResolver if your producers don't set headers.
    resolver = HeaderContractResolver(
        contract_loader=lambda ref: Validator.from_file(
            f"contracts/{ref.name}/{ref.version}.yaml"
        )
    )

    # 3. Where valid messages go, and where bad ones land.
    valid_sink = IteratorSink()
    dlq = MemoryDLQSink()

    processor = ContractStreamProcessor(
        source=source,
        valid_sink=valid_sink,
        dlq=dlq,
        resolver=resolver,
    )

    await processor.run()  # runs until cancelled


asyncio.run(main())

MemoryDLQSink is for testing — swap it for a Kafka, file, or S3-backed DLQ sink in production. The DLQReason enum (DECODE_ERROR, VALIDATION_FAILED, UNKNOWN_CONTRACT, …) is attached to every dead-lettered message so you can route or alert on it.

Streaming is experimental through the 0.x series — the public API is unlikely to shift but minor breaking changes may land between minor releases. See Streaming Validation for the deep dive on config, circuit-breaker tuning, and replay.