Metadata-Version: 2.4
Name: broker_core
Version: 0.2.1
Summary: Async-first message broker abstraction layer with pluggable backends.
Project-URL: homepage, https://example.com/broker_core
Project-URL: repository, https://example.com/broker_core.git
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: aiokafka>=0.8
Requires-Dist: pydantic>=2.0
Provides-Extra: nats
Requires-Dist: nats-py>=2.0; extra == "nats"
Provides-Extra: rabbitmq
Requires-Dist: aio-pika>=9.0; extra == "rabbitmq"

# 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`](https://example.com/db_core) — same fluent-builder pattern, same factory/env-var resolution, same extensibility model.

---

## Installation

```bash
# 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

```python
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

```python
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

```python
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:

```python
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

```python
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)

```python
# 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:

```python
# 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

```python
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

```python
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`)

```python
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

```python
# 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

```python
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)

```python
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)

```python
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

```python
# 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

```python
# 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.

```python
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

```python
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

```python
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:

```python
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

```python
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

```bash
# 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
