ouroboros_kafka 1.0.0

Production-ready Python wrapper around confluent-kafka with built-in auth management, delivery tracking and topic administration.

ouroboros_kafka

A production-ready Python wrapper around confluent-kafka providing high-level Producer, Consumer and AdminClient management with built-in authentication handling, automatic JSON serialization, delivery tracking and health checks.

---

Features

Area	Capabilities
Producer	Single / batch message production, automatic JSON serialization, delivery-report tracking, back-pressure handling, configurable flush timeouts, retry support
Consumer	Managed subscription, poll-based consumption, optional JSON deserialization, manual offset commit, conditional message search
AdminClient	Topic creation, listing, deletion, description, existence check via native Kafka protocol
Authentication	SASL (SCRAM-SHA-512/256, PLAIN, OAUTHBEARER), SSL/mTLS, auto-detection of protocol from bootstrap port
Health Check	Non-throwing cluster connectivity verification
Context Manager	with BaseKafka() as kafka: ... ensures all resources are released

---

Installation

pip install ouroboros_kafka

Dependencies

Package	Purpose
confluent-kafka==2.8.0	Core Kafka client (librdkafka-based)
ouroboros_json	JSON serialization / deserialization helpers

---

Quick Start

from ouroboros_kafka import BaseKafka, KafkaAuthConfig, SecurityProtocol

# 1. Configure authentication
auth = KafkaAuthConfig(
    bootstrap_servers="broker-1:9092,broker-2:9092",
    sasl_username="my-service",
    sasl_password="s3cr3t",
)

# 2. Instantiate the toolkit
kafka = BaseKafka()

# 3. (Optional) Create a topic via the native Kafka AdminClient API
kafka.create_topic_via_admin(
    "events.myservice",
    auth,
    num_partitions=6,
    replication_factor=3,
    config={"retention.ms": "604800000"},  # 7 days
)

# 4. Produce messages
kafka.create_producer(auth)
kafka.send_message("events.myservice", {"event": "user.created", "user_id": 42})
kafka.send_messages_batch("events.myservice", [
    {"event": "order.placed", "order_id": 1},
    {"event": "order.placed", "order_id": 2},
])
kafka.flush_producer()

# 5. Consume messages
kafka.create_consumer(auth, group_id="myservice-consumer")
kafka.subscribe("events.myservice")
messages = kafka.poll_messages(max_messages=10, deserialize_json=True)

# 6. Clean up
kafka.close()

---

Architecture

ouroboros_kafka/
+-- __init__.py          # Public API exports + __version__
+-- _config.py           # SecurityProtocol, SASLMechanism, KafkaAuthConfig
+-- _tracker.py          # DeliveryTracker (delivery-report accumulator)
+-- _producer.py         # ProducerMixin: create, send, batch, flush, retry
+-- _consumer.py         # ConsumerMixin: create, subscribe, poll, commit
+-- _admin.py            # AdminMixin: topics, health check
+-- base.py              # BaseKafka facade (composes all mixins) + legacy API
+-- exceptions.py        # Full exception hierarchy
+-- py.typed             # PEP 561 typed marker
tests/
+-- conftest.py          # Shared fixtures
+-- test_config.py       # Config / auth tests
+-- test_tracker.py      # DeliveryTracker tests
+-- test_producer.py     # Producer mixin tests
+-- test_consumer.py     # Consumer mixin tests
+-- test_admin.py        # Admin mixin tests
+-- test_base.py         # Facade / integration tests

---

Authentication

KafkaAuthConfig is an immutable dataclass that encapsulates all connection and security settings:

from pathlib import Path
from ouroboros_kafka import KafkaAuthConfig, SecurityProtocol, SASLMechanism

# SASL/SCRAM over SSL with mTLS
auth = KafkaAuthConfig(
    bootstrap_servers="kafka.prod.internal:9093",
    security_protocol=SecurityProtocol.SASL_SSL,
    sasl_mechanism=SASLMechanism.SCRAM_SHA_512,
    sasl_username="producer-svc",
    sasl_password="hunter2",
    ssl_ca_location=Path("/etc/kafka/ca.pem"),
    ssl_certificate_location=Path("/etc/kafka/client.pem"),
    ssl_key_location=Path("/etc/kafka/client-key.pem"),
)

The configuration is validated at connection time and secrets are never logged.

---

Context Manager

with BaseKafka() as kafka:
    kafka.create_producer(auth)
    kafka.send_message("topic", {"key": "value"}, flush=True)
# Producer, consumer and admin client are automatically closed here.

---

Producer

Creating a Producer

kafka = BaseKafka()
kafka.create_producer(auth, extra_config={"linger.ms": 100})

Sending Messages

# Single message (dict is auto-serialized to JSON)
kafka.send_message("topic", {"key": "value"})

# With key, headers, and specific partition
kafka.send_message(
    "topic",
    {"event": "click"},
    key="user-42",
    headers={"source": "web"},
    partition=0,
    flush=True,
)

# Batch of messages
kafka.send_messages_batch("topic", [
    {"event": "a"},
    {"event": "b"},
], key="shared-key", delay=0.1)

Producing with Retry

kafka.produce_with_retry(
    "topic",
    {"important": "data"},
    retries=5,
    retry_delay=2.0,
)

Delivery Tracking

kafka.create_producer(auth)
kafka.send_messages_batch("topic", messages)
kafka.flush_producer()

tracker = kafka.delivery_tracker
print(f"Delivered: {tracker.succeeded}, Failed: {tracker.failed}")
tracker.raise_on_failure()  # raises KafkaDeliveryError if any failed
tracker.reset()             # reset counters for the next batch

---

Consumer

kafka.create_consumer(
    auth,
    group_id="my-group",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
)
kafka.subscribe(["topic-a", "topic-b"])

# Poll up to 10 messages with JSON deserialization
messages = kafka.poll_messages(max_messages=10, timeout=2.0, deserialize_json=True)

# Search for the first message matching a condition
match = kafka.search_message(
    lambda msg: msg.get("testval_id") == "abc-123",
    timeout=30.0,
    deserialize_json=True,
)

# Search for multiple matching messages (e.g. wait until N events arrive)
matches = kafka.search_messages(
    lambda msg: msg.get("testval_id") == "abc-123",
    timeout=60.0,
    max_matches=5,
    deserialize_json=True,
)

# Manual commit
kafka.commit_offsets(asynchronous=False)

kafka.close_consumer()

---

Admin Client

Topic Operations

# Create
kafka.create_topic_via_admin(
    "events.orders",
    auth,
    num_partitions=12,
    replication_factor=3,
    config={"retention.ms": "86400000"},
)

# List all topics
topics = kafka.list_topics(auth)

# Check existence
if kafka.topic_exists("events.orders", auth):
    print("Topic exists")

# Describe (partitions, replicas, ISRs)
desc = kafka.describe_topic("events.orders", auth)
print(f"Partitions: {desc['partition_count']}")

# Delete
kafka.delete_topic("old-topic", auth)

---

Health Check

kafka = BaseKafka()
if kafka.health_check(auth, timeout=5.0):
    print("Cluster is reachable")
else:
    print("Cluster unreachable")

---

Legacy API

For backward compatibility a convenience method is available that auto-creates a producer:

kafka = BaseKafka()
kafka.send_kafka_message(
    bootstrap_servers="broker:9092",
    username="user",
    password="pass",
    topic="my-topic",
    message={"hello": "world"},
)

The security protocol is auto-detected from the bootstrap port (443, 9093, 30443 -> SASL_SSL; otherwise SASL_PLAINTEXT).

---

Exception Hierarchy

KafkaToolkitError
+-- KafkaConnectionError
+-- KafkaAuthenticationError
+-- KafkaTimeoutError
+-- KafkaProducerError
|   +-- KafkaSerializationError
|   +-- KafkaDeliveryError
+-- KafkaTopicError
    +-- KafkaTopicCreationError

All exceptions inherit from KafkaToolkitError for convenient broad catches.

---

API Reference

BaseKafka()

Method	Description
create_producer(auth, *, extra_config=None)	Create/replace the internal Producer
send_message(topic, message, *, key, headers, partition, on_delivery, flush)	Produce a single message
send_messages_batch(topic, messages, *, key, delay, flush)	Produce multiple messages
produce_with_retry(topic, message, *, key, headers, retries=3, retry_delay=1.0)	Produce with automatic retry on transient failures
flush_producer(timeout=30.0)	Flush the producer queue
close_producer()	Flush and dispose the producer
delivery_tracker	Access the DeliveryTracker for the current producer
create_consumer(auth, *, group_id, auto_offset_reset, enable_auto_commit, extra_config)	Create/replace the internal Consumer
subscribe(topics)	Subscribe the consumer to topic(s)
poll_messages(*, max_messages, timeout, deserialize_json)	Poll messages from the consumer
search_message(condition, *, timeout, poll_batch_size, deserialize_json)	Return the first message matching a condition, or None on timeout
search_messages(condition, *, timeout, max_matches, poll_batch_size, deserialize_json)	Collect up to max_matches messages matching a condition
commit_offsets(*, asynchronous=True)	Manually commit consumer offsets
close_consumer()	Close and dispose the consumer
create_admin_client(auth, *, extra_config)	Create/replace the internal AdminClient
list_topics(auth=None, *, timeout=10.0)	List all cluster topics
topic_exists(topic, auth=None, *, timeout=10.0)	Check if a topic exists
create_topic_via_admin(topic, auth, *, num_partitions, replication_factor, config, timeout)	Create a topic via AdminClient
delete_topic(topic, auth=None, *, timeout=30.0)	Delete a topic via AdminClient
describe_topic(topic, auth=None, *, timeout=10.0)	Get topic metadata (partitions, replicas, ISRs)
health_check(auth=None, *, timeout=10.0)	Non-throwing connectivity check
close()	Release all resources
send_kafka_message(...)	Legacy single-message API (auto-creates producer)

KafkaAuthConfig

Field	Type	Default
bootstrap_servers	str	(required)
security_protocol	SecurityProtocol	SASL_PLAINTEXT
sasl_mechanism	SASLMechanism | None	SCRAM_SHA_512
sasl_username	str | None	None
sasl_password	str | None	None
ssl_ca_location	Path | None	None
ssl_certificate_location	Path | None	None
ssl_key_location	Path | None	None
ssl_key_password	str | None	None
extra	dict	{}

Method	Description
to_confluent_config()	Build a confluent_kafka-compatible config dict
to_safe_string()	Human-readable representation (secrets masked)

DeliveryTracker

Property / Method	Description
succeeded	Number of successfully delivered messages
failed	Number of delivery failures
total	Total delivery reports received
errors	Copy of error message strings
raise_on_failure()	Raise KafkaDeliveryError if any delivery failed
reset()	Reset all counters and error lists

Enums

Enum	Values
SecurityProtocol	PLAINTEXT, SASL_PLAINTEXT, SASL_SSL, SSL
SASLMechanism	SCRAM_SHA_512, SCRAM_SHA_256, PLAIN, OAUTHBEARER

---

Development

pip install -e ".[dev]"

# Run tests
python -m pytest tests/ -v

# Lint
ruff check .

# Type check
mypy ouroboros_kafka/

---

License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Authors

Flavio Brandolini