Streaming — start here

PyCharter has two streaming surfaces, layered on the same BaseExtractor. Pick by what you need per message:

You want	Surface	Entry point
Validate every record against a contract as it arrives — header-based contract resolution, contract toggles (validate/coerce/annotate), per-message transforms, DLQ routing with a structured envelope + registry-outage retry, detect / shadow / enforce rollout modes, exactly_once delivery, windowed aggregation with durable checkpoints, circuit breaker, Prometheus metrics, worker CLI (stream validate / run --dry-run / list / inspect), Python one-liners (pycharter.run_stream)	Streaming validation engine (pycharter.kafka, pycharter stream run)	Streaming Validation guide
Feed a batch pipeline from a live source — SSE, WebSocket, Kafka, RabbitMQ, SQS, Redis Streams extractors yielding batches of records into ETL steps	Batch extractor surface (pipeline runtime)	Streaming and Messaging Extractors

No migration is required between them — the batch path keeps working unchanged if that is all you use.

Per-record contract validation

• Streaming Validation guide — library API (pycharter.kafka.consume), operating modes, DLQ envelope, delivery semantics.
• Streaming Contract Validation tutorial — end-to-end Kafka walkthrough with a worker config.
• Streaming Worker Config Schema — every pycharter stream run YAML key.
• Consume a Kafka topic (cookbook) — the 30-second version.
• Observability: Prometheus metrics ship with the engine (see Emit a quality score to Prometheus); the [otel] extra adds per-record OpenTelemetry spans parented into the producer's trace.

Batch extractors from live sources

• Streaming and Messaging Extractors — SSE / WebSocket / file-watcher / Kafka / RabbitMQ / SQS, batch controls, acknowledgment semantics.
• Incremental Extraction — resumable cursors for append-only sources.
• The batch pipeline rejects an unbounded streaming source unless max_records / max_batches is set — continuous consumption belongs to pycharter stream run.