broker_core

Async-first message broker abstraction layer with pluggable backends.
Currently supports Kafka via aiokafka; NATS and RabbitMQ are designed as future drop-ins.

Built as the messaging counterpart to db_core — same fluent-builder pattern, same factory/env-var resolution, same extensibility model.

Installation

# Core library (no backend drivers)
pip install broker_core

# With Kafka support
pip install broker_core[kafka]

# With all backends (future)
pip install broker_core[kafka,nats,rabbitmq]

# Development install (editable, from source)
pip install -e ".[kafka]"

Requirements: Python ≥ 3.11, a running Kafka broker (for Kafka features).

Quick Start

import asyncio
from broker_core import create_kafka_client

async def main():
    # Create and connect
    client = create_kafka_client(bootstrap_servers="localhost:9092")
    await client.connect()

    # Publish
    await client.topic("orders") \
        .key("order-123") \
        .payload({"id": 123, "status": "created", "amount": 99.95}) \
        .headers({"source": "web"}) \
        .publish()

    # Subscribe (async iterator)
    sub = await client.topic("orders") \
        .group_id("worker") \
        .from_beginning() \
        .subscribe()

    async with sub:
        async for msg in sub:
            print(f"Received: {msg.payload}")
            await msg.ack()
            break  # consume just one

    await client.disconnect()

asyncio.run(main())

Client Creation

From Environment Variables

from broker_core import create_kafka_client

# Reads from KAFKA_BOOTSTRAP_SERVERS env var
client = create_kafka_client()

# Or get_kafka_client() — identical alias
from broker_core import get_kafka_client
client = get_kafka_client()

Environment variables checked (in priority order):

Explicit Configuration

from broker_core import create_kafka_client

client = create_kafka_client(
    bootstrap_servers="kafka1:9092,kafka2:9092",
    client_id="my-service",
    group_id="default-consumer",
)

Full KafkaConfig Object

For complete control — producer settings, consumer defaults, security:

from broker_core import KafkaClient, KafkaConfig

config = KafkaConfig(
    bootstrap_servers="broker:9093",
    client_id="production-svc",
    group_id="main-consumer",

    # Producer
    acks="all",                     # 0, 1, or "all"
    compression_type="snappy",      # gzip, snappy, lz4, zstd
    max_request_size=2_097_152,     # 2 MB

    # Consumer defaults (overridable per subscription)
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    max_poll_records=100,
    session_timeout_ms=30000,

    # Security
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",
    sasl_username="app-user",
    sasl_password="app-secret",
)

client = KafkaClient(config)
await client.connect()

Publishing Messages

Fluent Builder

result = await client.topic("orders") \
    .key("order-123") \
    .payload({"id": 123, "status": "created"}) \
    .headers({"source": "web", "region": "us-east"}) \
    .publish()

print(f"Partition: {result.partition}, Offset: {result.offset}")

Builder Methods (Publish Chain)

Convenience Methods (Skip the Builder)

# Quick publish
await client.publish("alerts", {"msg": "disk full"}, key="alert-1")

# With headers
await client.publish(
    "events",
    {"type": "user.signup"},
    key="user-42",
    headers={"source": "api", "version": "2"},
)

# emit() is an alias
await client.emit("logs", {"level": "info", "message": "started"})

Publishing Without a Key

Omitting .key() tells Kafka to round-robin across partitions — best for independent events where ordering doesn't matter:

# Each publish may land in a different partition
await client.topic("page-views").payload({"page": "/home"}).publish()
await client.topic("page-views").payload({"page": "/cart"}).publish()
await client.topic("page-views").payload({"page": "/checkout"}).publish()

Adding Single Headers

await client.topic("events") \
    .header("trace-id", "abc-123") \
    .header("content-type", "application/json") \
    .payload({"event": "payment.received"}) \
    .publish()

Subscribing to Messages

Async Iterator Mode

sub = await client.topic("orders") \
    .group_id("order-processor") \
    .from_beginning() \
    .subscribe()

async with sub:
    async for msg in sub:
        print(f"Order: {msg.payload}")
        await msg.ack()          # commit offset
        # or await msg.nack()    # signal failure, seek back

Callback Mode (listen)

async def handle_order(msg):
    await process(msg.payload)
    await msg.ack()

sub = await client.topic("orders") \
    .group_id("order-worker") \
    .enable_auto_commit(False) \
    .max_poll_records(10) \
    .listen(handle_order)

# ... application runs ...

await sub.stop()  # graceful shutdown

Consumer Configuration (Subscribe Chain)

Convenience Subscribe

# Skip the builder
sub = await client.subscribe(
    "orders",
    group_id="worker",
    auto_offset_reset="earliest",
)
async with sub:
    async for msg in sub:
        await msg.ack()

# Convenience listen
sub = await client.listen("orders", handle_order, group_id="worker")

Manual Offset Control

sub = await client.topic("orders") \
    .group_id("worker") \
    .enable_auto_commit(False) \
    .subscribe()

async with sub:
    async for msg in sub:
        try:
            await process(msg.payload)
            await msg.ack()       # commit on success
        except Exception:
            await msg.nack()      # seek back for retry

    # Manual commit at any point
    await sub.commit()

    # Seek to beginning/end
    await sub.seek_to_beginning()
    await sub.seek_to_end()

MessageResult (Publish Acknowledgment)

result = await client.topic("orders").payload({...}).publish()

result.topic       # "orders"
result.partition   # 0
result.offset      # 42
result.timestamp   # 1718053200000 (ms)
result.success     # True
result.failed()    # False

ReceivedMessage (Consumed Message)

async for msg in subscription:
    msg.topic         # "orders"
    msg.partition     # 0
    msg.offset        # 42
    msg.timestamp     # 1718053200000
    msg.key           # "order-123" or None
    msg.payload       # {"id": 123, "status": "created"}
    msg.headers       # {"source": "web"}

    await msg.ack()   # commit offset
    await msg.nack()  # signal failure

Subscription Control

# Async context manager — auto-stops on exit
async with sub:
    async for msg in sub:
        ...

# Manual lifecycle
sub = await client.topic("orders").group_id("w").subscribe()
async for msg in sub:
    ...
await sub.stop()

# Seek operations
await sub.seek_to_beginning()
await sub.seek_to_end()
await sub.commit()

Introspection & Debugging

# Inspect the built message before publishing
builder = client.topic("orders").key("k").payload({"x": 1})
print(builder.to_dict())   # {'topic': 'orders', 'key': 'k', 'payload': ...}
print(builder.debug())     # Pretty-printed JSON

# Inspect a Message object directly
from broker_core import Message
msg = Message(topic="t", key="k", payload={"a": 1})
print(msg.debug())
print(msg.to_dict())

Complete Examples

Long-Running Service

import asyncio
import signal
from broker_core import create_kafka_client, ReceivedMessage

async def handle_payment(msg: ReceivedMessage):
    payment = msg.payload
    print(f"Processing payment {payment['id']}: ${payment['amount']}")
    await msg.ack()

async def main():
    client = create_kafka_client(client_id="payment-service")
    await client.connect()

    sub = await client.topic("payments") \
        .group_id("payment-processor") \
        .from_beginning() \
        .listen(handle_payment)

    # Wait for shutdown signal
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop_event.set)

    await stop_event.wait()
    print("Shutting down...")
    await sub.stop()
    await client.disconnect()

asyncio.run(main())

Fan-Out: One Publisher, Multiple Consumer Groups

async def consumer_a(client):
    sub = await client.topic("events").group_id("group-a").from_beginning().subscribe()
    async with sub:
        async for msg in sub:
            print(f"A got: {msg.payload}")
            await msg.ack()

async def consumer_b(client):
    sub = await client.topic("events").group_id("group-b").from_beginning().subscribe()
    async with sub:
        async for msg in sub:
            print(f"B got: {msg.payload}")
            await msg.ack()

async def main():
    client = create_kafka_client()

    # Publish
    await client.topic("events").payload({"msg": "hello"}).publish()

    # Both groups receive the message independently
    await asyncio.gather(consumer_a(client), consumer_b(client))

Using the Same Builder Instance Repeatedly

orders = client.topic("orders")

# Publish three messages — builder auto-resets after each publish
await orders.key("A").payload({"id": "A"}).publish()
await orders.key("B").payload({"id": "B"}).publish()
await orders.key("C").payload({"id": "C"}).publish()

# Consumer config is sticky — set once, reuse
orders.group_id("worker").from_beginning()

sub1 = await orders.subscribe()
# ... consumer config preserved
sub2 = await orders.subscribe()

API Reference

Top-Level Imports

from broker_core import (
    # Client
    KafkaClient,
    KafkaConfig,

    # Factory
    create_kafka_client,
    get_kafka_client,
    create_nats_client,       # NotImplementedError (future)
    create_rabbitmq_client,   # NotImplementedError (future)

    # Message model
    MessageBuilder,
    Message,

    # Result types
    MessageResult,
    Subscription,
    ReceivedMessage,

    # Routing
    TopicPattern,
    PartitionTarget,
)

Class Hierarchy

BrokerClient (ABC)
├── connect() / disconnect()
├── topic(name) → MessageBuilder
├── _publish(message) → MessageResult
├── _subscribe(message, builder) → Subscription
├── publish() / emit() / subscribe() / listen()   # convenience
├── raw_message() / debug_message()               # debug
│
└── KafkaClient (BrokerClient)
    ├── config: KafkaConfig
    ├── _publish() → producer.send_and_wait()
    └── _subscribe() → AIOKafkaConsumer + consume loop

MessageBuilder
├── .key() .payload() .headers() .header() .partition()
├── .group_id() .client_id() .auto_offset_reset() .from_beginning()
├── .enable_auto_commit() .auto_commit_interval_ms()
├── .max_poll_records() .session_timeout_ms()
├── .to_dict() .debug()
├── .publish() / .emit()
└── .subscribe() / .listen()

Subscription
├── __aenter__ / __aexit__  (async context manager)
├── __aiter__ / __anext__   (async iterator)
├── stop() / commit()
└── seek_to_beginning() / seek_to_end()

MessageResult      # topic, partition, offset, timestamp, success
ReceivedMessage    # topic, partition, offset, timestamp, key, payload, headers
KafkaConfig        # All Kafka settings in one dataclass

Testing

# Run all tests (requires Docker for Kafka)
./run_tests.fish

# Unit tests only (no Kafka needed)
./run_tests.fish --unit

# Kafka integration tests only
./run_tests.fish --kafka

# Or with pytest directly
pytest tests/ -v                # all
pytest tests/ -v -m "not kafka"  # unit only
pytest tests/ -v -m "kafka"      # kafka only

Adding a New Backend

1. Create clients/<name>.py — extend BrokerClient, implement connect, disconnect, topic, _publish, _subscribe
2. Add factory function in clients/factory.py
3. Export from __init__.py
4. Add optional dependency in pyproject.toml
5. Add tests in tests/test_<name>.py and fixture in tests/conftest.py

License

MIT