base.kafka package

Public Kafka implementation package.

class base.kafka.BufferedKafkaProduceHandler

Bases: SimpleKafkaProduceHandler

Asynchronous producer with bounded local backpressure.

produce(topic: str, data: str, key: None | str = None) → None

class base.kafka.ConsumedKafkaMessage(key: str | None, value: str | None, topic: str, partition: int, offset: int, raw_message: object | None = None)

Bases: object

Decoded Kafka record plus the source offset required for EoS.

key: str | None

offset: int

partition: int

raw_message: object | None = None

topic: str

value: str | None

class base.kafka.ExactlyOnceKafkaConsumeHandler(topics: str | list[str])

Bases: KafkaConsumeHandler

Consumer that exposes only committed transactional records.

consume() → tuple[str | None, str | None, str | None]

class base.kafka.ExactlyOnceKafkaProduceHandler

Bases: KafkaProduceHandler

Transactional producer that atomically commits outputs and input offsets.

commit_transaction_with_retry(max_retries: int = 3, retry_interval_ms: int = 1000) → None

produce(topic: str, data: str, key: None | str = None) → None

transaction_batch(consumer: KafkaConsumeHandler, consumed_messages: Sequence[ConsumedKafkaMessage]) → Iterator[None]

Collect outputs and commit them with source offsets on clean exit.

class base.kafka.KafkaConsumeHandler(topics: str | list[str])

Bases: KafkaSerializationMixin, KafkaHandler

Common connection, batching, decoding, and offset behavior.

commit(consumed_messages: Sequence[ConsumedKafkaMessage] | None = None) → None

Commit an explicit batch or the latest record from consume.

abstract consume(*args, **kwargs)

consume_batch(max_messages: int | None = None, timeout_ms: int | None = None) → list[ConsumedKafkaMessage]

Fetch a bounded group of records without committing offsets.

group_metadata()

static offsets_for(consumed_messages: Sequence[ConsumedKafkaMessage]) → list[TopicPartition]

Return the highest processed next offset for each source partition.

class base.kafka.KafkaHandler

Bases: object

Base class shared by producer and consumer wrappers.

exception base.kafka.KafkaMessageFetchException

Bases: Exception

Raised when Kafka returns a permanent consumer error.

class base.kafka.KafkaProduceHandler(conf)

Bases: KafkaHandler

Common lifecycle and retry behavior for Kafka producers.

abstract produce(*args, **kwargs)

class base.kafka.KafkaProduceRecord(topic: str, data: str, key: str | None = None)

Bases: object

One output record queued for a Kafka transaction.

data: str

key: str | None = None

topic: str

class base.kafka.KafkaTopicManager(admin_client: AdminClient)

Bases: object

Reconcile a set of topics with the configured partition policy.

ensure(topics: str | list[str], target_partitions: int | None = None, replication_factor: int | None = None, auto_expand_partitions: bool | None = None) → dict[str, int]

class base.kafka.SimpleKafkaConsumeHandler(topics: str | list[str])

Bases: KafkaConsumeHandler

Consumer without transactional read isolation.

consume() → tuple[str | None, str | None, str | None]

class base.kafka.SimpleKafkaProduceHandler

Bases: KafkaProduceHandler

Synchronous Kafka producer without transactional semantics.

produce(topic: str, data: str, key: None | str = None) → None

exception base.kafka.TooManyFailedAttemptsError

Bases: Exception

Raised when required Kafka topics never become available.

base.kafka.build_consumer_group_id(topics: str | list[str]) → str

base.kafka.ensure_topics(admin_client: AdminClient, topics: str | list[str], target_partitions: int | None = None, replication_factor: int | None = None, auto_expand_partitions: bool | None = None) → dict[str, int]

Preserve the existing functional API around the topic manager.

Submodules

base.kafka.client module

Common base type for Kafka client wrappers.

class base.kafka.client.KafkaHandler

Bases: object

Base class shared by producer and consumer wrappers.

base.kafka.config module

Process-level Kafka configuration shared by all Kafka clients.

base.kafka.config.bootstrap_servers() → str

Return the configured brokers in confluent-kafka format.

base.kafka.consumer module

Kafka consumer hierarchy and offset management.

class base.kafka.consumer.ExactlyOnceKafkaConsumeHandler(topics: str | list[str])

Bases: KafkaConsumeHandler

Consumer that exposes only committed transactional records.

consume() → tuple[str | None, str | None, str | None]

class base.kafka.consumer.KafkaConsumeHandler(topics: str | list[str])

Bases: KafkaSerializationMixin, KafkaHandler

Common connection, batching, decoding, and offset behavior.

commit(consumed_messages: Sequence[ConsumedKafkaMessage] | None = None) → None

Commit an explicit batch or the latest record from consume.

abstract consume(*args, **kwargs)

consume_batch(max_messages: int | None = None, timeout_ms: int | None = None) → list[ConsumedKafkaMessage]

Fetch a bounded group of records without committing offsets.

group_metadata()

static offsets_for(consumed_messages: Sequence[ConsumedKafkaMessage]) → list[TopicPartition]

Return the highest processed next offset for each source partition.

class base.kafka.consumer.SimpleKafkaConsumeHandler(topics: str | list[str])

Bases: KafkaConsumeHandler

Consumer without transactional read isolation.

consume() → tuple[str | None, str | None, str | None]

base.kafka.errors module

Kafka wrapper exceptions.

exception base.kafka.errors.KafkaMessageFetchException

Bases: Exception

Raised when Kafka returns a permanent consumer error.

exception base.kafka.errors.TooManyFailedAttemptsError

Bases: Exception

Raised when required Kafka topics never become available.

base.kafka.producer module

Kafka producer hierarchy for simple, buffered, and EoS delivery.

class base.kafka.producer.BufferedKafkaProduceHandler

Bases: SimpleKafkaProduceHandler

Asynchronous producer with bounded local backpressure.

produce(topic: str, data: str, key: None | str = None) → None

class base.kafka.producer.ExactlyOnceKafkaProduceHandler

Bases: KafkaProduceHandler

Transactional producer that atomically commits outputs and input offsets.

commit_transaction_with_retry(max_retries: int = 3, retry_interval_ms: int = 1000) → None

produce(topic: str, data: str, key: None | str = None) → None

transaction_batch(consumer: KafkaConsumeHandler, consumed_messages: Sequence[ConsumedKafkaMessage]) → Iterator[None]

Collect outputs and commit them with source offsets on clean exit.

class base.kafka.producer.KafkaProduceHandler(conf)

Bases: KafkaHandler

Common lifecycle and retry behavior for Kafka producers.

abstract produce(*args, **kwargs)

class base.kafka.producer.SimpleKafkaProduceHandler

Bases: KafkaProduceHandler

Synchronous Kafka producer without transactional semantics.

produce(topic: str, data: str, key: None | str = None) → None

base.kafka.records module

Value objects passed between Kafka consumers and producers.

class base.kafka.records.ConsumedKafkaMessage(key: str | None, value: str | None, topic: str, partition: int, offset: int, raw_message: object | None = None)

Bases: object

Decoded Kafka record plus the source offset required for EoS.

key: str | None

offset: int

partition: int

raw_message: object | None = None

topic: str

value: str | None

class base.kafka.records.KafkaProduceRecord(topic: str, data: str, key: str | None = None)

Bases: object

One output record queued for a Kafka transaction.

data: str

key: str | None = None

topic: str

base.kafka.resilience module

Kafka-specific transient error classification.

base.kafka.resilience.is_retriable_kafka_error(error) → bool

base.kafka.resilience.is_retriable_kafka_exception(exception: Exception) → bool

base.kafka.serialization module

Deserialization behavior shared by Kafka consumer implementations.

class base.kafka.serialization.KafkaSerializationMixin

Bases: object

Decode consumed Kafka values while leaving transport logic separate.

consume_as_json(source_message: ConsumedKafkaMessage | None = None) → tuple[str | None, dict]

consume_as_object(source_message: ConsumedKafkaMessage | None = None) → tuple[None | str, Batch]

base.kafka.topics module

Kafka topic configuration, provisioning, and consumer-group naming.

class base.kafka.topics.KafkaTopicManager(admin_client: AdminClient)

Bases: object

Reconcile a set of topics with the configured partition policy.

ensure(topics: str | list[str], target_partitions: int | None = None, replication_factor: int | None = None, auto_expand_partitions: bool | None = None) → dict[str, int]

base.kafka.topics.build_consumer_group_id(topics: str | list[str]) → str

base.kafka.topics.ensure_topics(admin_client: AdminClient, topics: str | list[str], target_partitions: int | None = None, replication_factor: int | None = None, auto_expand_partitions: bool | None = None) → dict[str, int]

Preserve the existing functional API around the topic manager.

base.kafka.topics.normalize_topics(topics: str | list[str]) → list[str]

base.kafka.topics.topic_partition_count(cluster_metadata, topic: str) → int | None