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

# Kafka support is built in — aiokafka + pydantic are core dependencies
pip install broker_core

# Future backends (optional extras — clients not yet implemented)
pip install broker_core[nats]
pip install broker_core[rabbitmq]

# Development install (editable, from source)
pip install -e .

Requirements: Python ≥ 3.11, a running Kafka broker.

---

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):

Variable	Purpose
KAFKA_BOOTSTRAP_SERVERS	Comma-separated broker addresses
KAFKA_URL	Fallback URL
KAFKA_HOST + KAFKA_PORT	Host + port (default port 9092)
KAFKA_CLIENT_ID	Client identifier
KAFKA_GROUP_ID	Default consumer group
KAFKA_AUTO_OFFSET_RESET	earliest or latest
KAFKA_SECURITY_PROTOCOL	PLAINTEXT, SASL_SSL, etc.
KAFKA_SASL_MECHANISM	PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
KAFKA_SASL_USERNAME	SASL username
KAFKA_SASL_PASSWORD	SASL password

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)

Method	Description
.key(key)	Message key — determines partition routing
.payload(data)	JSON-serialisable dictionary
.headers(headers)	Metadata dict (merged with previous)
.header(key, value)	Add a single header
.partition(n)	Target a specific partition
.publish()	Terminal — send and return MessageResult
.emit()	Alias for.publish()

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)

| Method | Description | | ------------------------------ | ----------------------------------------------------- | ------------------------------------ | | .group_id(gid) | Consumer group identifier | | .client_id(cid) | Per-consumer client ID override | | .auto_offset_reset("earliest" | "latest") | Where to start when no offset exists | | .from_beginning() | Sugar for.auto_offset_reset("earliest") | | .enable_auto_commit(bool) | Auto-commit offsets (default:True) | | .auto_commit_interval_ms(ms) | Interval between auto-commits | | .max_poll_records(n) | Max records per poll | | .session_timeout_ms(ms) | Consumer session timeout | | .subscribe() | Terminal — return Subscription (async iterator) | | .listen(callback) | Terminal — callback-based consumer |

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())

---

Topics & Payload Models

broker_core/topics/ is the single source of truth for Kafka topic names and payload schemas.

from broker_core import Topic, TelegramMessagePayload, TOPIC_PAYLOAD_MAP, get_payload_model

# Use the enum instead of hardcoded strings
Topic.TELEGRAM_MESSAGE_RAW_READY            # "telegram_message_raw_ready"
Topic.WEB_RAW_READY                         # "web.raw.ready"

# Look up the expected payload model for a topic (None for unknown topics)
model = get_payload_model("telegram_message_raw_ready")  # TelegramMessagePayload

# Validate an incoming payload
parsed = TelegramMessagePayload.model_validate(raw_dict)

Topics

• Telegram (telegram_{domain}_{status}) — domains raw | parse | analyze | media | persistence | metrics | events, statuses ready | persisted | failed | dead, plus telegram_backfill_job_ready
• Web (dotted names) — web.url.ready, web.raw.retry, web.raw.dlq, web.raw.ready, web.raw.persisted, web.discovery.ready, web.parsed.ready, web.parsed.persisted, codal.web.raw.persisted, codal.web.parsed.persisted, codal.media.ready, web.crawl.metrics
• Web downloader — web.download.ready, web.download.retry, web.download.dlq, web.media.ready, web.media.metrics
• Bale — bale_message_{domain}_{status} mirroring the Telegram/parse/media/metrics/events domains, plus bale_channel_sync_ready
• Entities (new-analytics) — web.entities.extracted, web.entities.enriched

Payload models

Model	Topics
TelegramMessagePayload	telegram_message_raw_ready, telegram_message_raw_persisted
TelegramParsedPayload (wraps TelegramParsedMessagePayload + MessageReactionPayload[] + MessageDetailPayload[])	telegram_message_parse_ready, telegram_message_parse_persisted, telegram_message_analyze_persisted
TelegramAnalyzeReadyPayload	telegram_message_analyze_ready
MediaFilePayload	telegram_message_media_ready, telegram_message_media_persisted
TelegramCrawlMetricPayload	telegram_message_metrics_ready
CrawlerEventPayload	telegram_message_events_ready
PipelineErrorPayload	all*.failed / *.dead topics
BackfillJobPayload	telegram_backfill_job_ready
UrlReadyPayload / RawReadyPayload / DiscoveryReadyPayload / CrawlMetricPayload / ParsedReadyPayload / ParsedPersistedPayload	web pipeline topics
WebDownloadReadyPayload / WebMediaReadyPayload / WebDownloadMetricPayload	downloader topics
EntityExtractedPayload / EntityEnrichedPayload (EntityPayload[])	web.entities.extracted, web.entities.enriched

All models use extra="allow" (forward-compatible wire format). Telegram/web/backfill models prefer optional fields; metrics/events/error models (TelegramCrawlMetricPayload, CrawlerEventPayload, PipelineErrorPayload) use from_attributes=True.

Datetimes on the wire must be ISO 8601 strings (e.g. "2026-06-17T12:00:00+00:00"), never datetime objects — they crash json.dumps(). Metrics/events/error models accept datetime | str for ORM-style construction; serialize with datetime.now(timezone.utc).isoformat() before publishing.

---

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

client.topic("orders") returns a new MessageBuilder each call — keep the returned reference to reuse one builder (and its sticky consumer config) across operations:

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)
    get_nats_client,          # NotImplementedError (future)
    create_rabbitmq_client,   # NotImplementedError (future)
    get_rabbitmq_client,      # NotImplementedError (future)

    # Message model
    MessageBuilder,
    Message,

    # Result types
    MessageResult,
    Subscription,
    ReceivedMessage,

    # Routing
    TopicPattern,
    PartitionTarget,

    # Topics & payload models
    Topic,
    TOPIC_PAYLOAD_MAP,
    get_payload_model,
    TelegramMessagePayload,
    TelegramParsedPayload,
    TelegramParsedMessagePayload,
    MessageDetailPayload,
    MessageReactionPayload,
    TelegramAnalyzeReadyPayload,
    MediaFilePayload,
    TelegramCrawlMetricPayload,
    CrawlerEventPayload,
    PipelineErrorPayload,
    BackfillJobPayload,
    UrlReadyPayload,
    RawReadyPayload,
    DiscoveryReadyPayload,
    CrawlMetricPayload,
    ParsedReadyPayload,
    ParsedPersistedPayload,
    WebDownloadReadyPayload,
    WebDownloadMetricPayload,
)

Class Hierarchy

BrokerClient (ABC)
├── connect() / disconnect()
├── topic(name) → MessageBuilder   # fresh instance per call — hold the reference to reuse
├── _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

Topic (StrEnum)             # 40 topic name constants (telegram_* + web.* + codal.*)
TOPIC_PAYLOAD_MAP           # Topic → Pydantic payload model registry
get_payload_model(topic)    # Runtime model lookup (None for unknown topics)

---

Testing

# Run all tests (requires Docker for Kafka)
./run_tests.fish          # or ./run_tests.sh (Bash/Zsh)

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

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

The scripts create the virtual environment, install dependencies, start Kafka via the shared infrastructure stack, wait for readiness, then run pytest tests/ -v with KAFKA_BOOTSTRAP_SERVERS=localhost:9094 (host external listener). Kafka integration tests are marked kafka (module-level pytestmark) and are skipped in unit-only runs.

Adding a New Backend

1. Create clients/<name>.py — extend BrokerClient and implement connect(), disconnect(), topic(name) (return a fresh MessageBuilder(self, name) per call), _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