warren.pubsub.kafka
Kafka pubsub layer.
The top-level module exposes only the protocol-level configs —
pure-data Pydantic models that describe the Kafka topology and consumer
settings. They have no dependency on any specific Python client library
and can be loaded, serialized, and reused freely (importing this module
does not require aiokafka).
Concrete client implementations live in sub-packages:
aiokafka—aiokafka-based managers and helpers (requires thewarren[kafka]extra).
To use the aiokafka implementation, import explicitly from that
sub-package, e.g.:
from warren.pubsub.kafka.aiokafka import (
KafkaConnectionManager,
KafkaConsumerManager,
KafkaPublisher,
)Modules
Classes
KafkaConnectionConfig
modelKafkaConnectionConfigCluster-level parameters shared by all Kafka clients.
The SSL fields are file paths only — the ssl.SSLContext is
built in the aiokafka implementation sub-package, keeping this
model pure-data.
attributebootstrap_serverslist[str]= (lambda: ['localhost:9092'])()Broker addresses to seed cluster discovery.
attributesecurity_protocolLiteral['PLAINTEXT', 'SSL']= 'PLAINTEXT'"PLAINTEXT" or "SSL".
attributessl_cafilestr | None= NonePath to the CA-certificate file (SSL only).
attributessl_certfilestr | None= NonePath to the client-certificate file (SSL, mutual-TLS only).
attributessl_keyfilestr | None= NonePath to the client private-key file (SSL, mutual-TLS only).
attributeclient_idstr | None= NoneClient identifier reported to the broker; when None, aiokafka uses its own default.
KafkaConsumerConfig
modelKafkaConsumerConfigConsumer-group parameters for a single consumer manager.
attributeconcurrencyint | None= NoneMust be None or 1, so that offsets are committed in order. One handler runs at a time.
attributegroup_idstr | None= NoneConsumer group name. When None, the consumer manager derives "<topic>.<worker_type>" — one group per worker type on the topic (fanout parity). Set explicitly on platforms with pre-assigned group names.
attributeauto_offset_resetLiteral['earliest', 'latest']= 'earliest'Where a new group starts reading. Defaults to "earliest" so a freshly created group drains the existing backlog instead of skipping it (Kafka's own default is "latest").
attributemax_poll_interval_msint= 600000Max time between polls before the broker evicts the consumer from the group.
attributesession_timeout_msint= 45000Heartbeat session timeout.
attributehandler_timeout_secondsfloat | None= NoneTime limit for one handler call. A handler that times out is a soft failure. /live fails when a handler runs for twice this value. None means no limit.
attributeon_shutdown_timeoutfloat= 30.0Seconds to wait for the in-flight message during shutdown (mirrors RMQConsumerConfig).
KafkaConsumerManagerConfig
modelKafkaConsumerManagerConfigBundle of the topic and consumer config for one consumer manager.
attributetopicKafkaTopicConfigThe topic to consume from (and optionally create).
attributeconsumerKafkaConsumerConfigConsumer-group parameters for this manager.
KafkaTopicConfig
modelKafkaTopicConfigA Kafka topic — plays the role of a fanout exchange.
attributenamestrTopic name.
attributenum_partitionsint= 6Partition count, used only when creating.
attributereplication_factorint= 1Replication factor, used only when creating.
attributecreate_if_missingbool= FalseCreate the topic if it does not exist. Enable locally; disable on platforms where topics are provisioned out-of-band (e.g. via a console).