aiokafka-foundation-kit

Async Kafka foundation library — settings, client factories, lifecycle helpers, topic management and DI wiring on top of aiokafka.

Features

• AsyncIO-first — everything is built on aiokafka; there is no synchronous mirror
• Protocol-based config — every entry point takes a typing.Protocol, so a Pydantic model, a dataclass or a stub all work
• Pydantic settings — BaseKafkaProducerSettings / BaseKafkaConsumerSettings with validation, via the models extra
• Lifecycle helpers — producer_lifecycle / consumer_lifecycle own start and stop, including on failure
• Topic management — ensure_topics_async creates topics idempotently before a producer starts
• Health checks — check_kafka_health_async probes broker reachability
• JSON serialisation — dumps_bytes / loads_bytes, with transparent orjson acceleration
• DI integration — Dishka providers and dependency-injector containers
• OpenTelemetry — instrument_aiokafka wires the aiokafka instrumentor in one call
• Type-safe — full annotations, checked with mypy

Installation

# Core functionality
pip install aiokafka-foundation-kit

# With all optional dependencies
pip install aiokafka-foundation-kit[all]

# Or specific features
pip install aiokafka-foundation-kit[models,orjson,dishka,dependency-injector,telemetry]

Quick Start

Producer

from aiokafka_foundation_kit import TopicConfig, producer_lifecycle
from aiokafka_foundation_kit.contrib.models import BaseKafkaProducerSettings

settings = BaseKafkaProducerSettings(bootstrap_servers="localhost:9092")
topics = [TopicConfig(name="events", num_partitions=3, replication_factor=1)]

async with producer_lifecycle(settings, topics=topics, auto_create_topics=True) as producer:
    # dicts are serialised to JSON; keys must already be bytes
    await producer.send_and_wait("events", {"event": "user_created", "user_id": "123"})

producer_lifecycle yields aiokafka's own AIOKafkaProducer, started, and stops it when the block exits — including when the block raises.

Consumer

from aiokafka_foundation_kit import consumer_lifecycle
from aiokafka_foundation_kit.contrib.models import BaseKafkaConsumerSettings

settings = BaseKafkaConsumerSettings(
    bootstrap_servers="localhost:9092",
    group_id="my-group",
)

async with consumer_lifecycle(settings, topics=("events",)) as consumer:
    async for message in consumer:
        print(message.value)   # already decoded from JSON
        await consumer.commit()

enable_auto_commit defaults to False, so commit after processing to get at-least-once delivery. The async for loop is yours to write: this library manages the client, not the message flow.

Documentation

Full documentation at bedrock-python.github.io/aiokafka-foundation-kit.

For AI agents is the whole library on one page — the API surface, the rules that break code when they are broken, the mistakes models make, and a map of which page to fetch for the rest — written to be handed to a coding assistant.

Development

# Install dependencies
uv sync --group dev --all-extras

# Lint, format check and mypy
make check

# Run tests
make test-unit  # unit tests, no Docker required
make test       # unit + integration tests, requires Docker

License

Apache 2.0 — see LICENSE for details.