Metadata-Version: 2.4
Name: fastkafka2
Version: 0.6.1
Summary: Production-oriented async Kafka framework with bounded backpressure and fail-closed validation
Author: Ruslan Kiradiev
License-Expression: MIT
Keywords: kafka,async,fastapi,pydantic,microservices,messaging,event-driven
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: Internet
Classifier: Framework :: AsyncIO
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic<3,>=2.6
Requires-Dist: confluent-kafka<3,>=2.6
Requires-Dist: typing-extensions>=4.0.0
Requires-Dist: orjson<4,>=3.9
Provides-Extra: dev
Requires-Dist: pytest<10,>=9.0.3; extra == "dev"
Requires-Dist: pytest-asyncio<2,>=0.23; extra == "dev"
Requires-Dist: pytest-cov<7,>=5; extra == "dev"
Requires-Dist: ruff<1,>=0.9; extra == "dev"
Requires-Dist: mypy<2,>=1.11; extra == "dev"
Requires-Dist: bandit<2,>=1.7; extra == "dev"
Requires-Dist: pip-audit<3,>=2.7; extra == "dev"
Requires-Dist: build<2,>=1.2; extra == "dev"
Requires-Dist: twine<7,>=5; extra == "dev"
Dynamic: license-file

# fastkafka2

`fastkafka2` is an async, Pydantic-based Kafka framework built on
`confluent-kafka`/librdkafka. Version 0.6 is designed for services that need bounded
memory, explicit transport security, per-partition ordering, and at-least-once delivery.

## Production contract

- TLS/SASL configuration is mandatory and cannot be downgraded through raw client options.
- `Plaintext()` is available only as an explicit choice for local development.
- Consumer group IDs are explicit deployment identities; none are derived from broker URLs.
- Auto-commit and auto-offset-store are disabled. A synchronous broker-acknowledged commit
  advances only after ordered processing succeeds.
- Failed commits remain pending and are retried.
- Processing is ordered within a partition and concurrent across partitions.
- Both message count and total buffered bytes are bounded. Handler concurrency is globally
  bounded to protect databases and downstream APIs.
- Producer delivery is broker-acknowledged, idempotence is enabled, and batches are bounded.
- JSON/Pydantic validation is fail-closed. A handler must declare a concrete Pydantic data
  model.
- Topic and DLT auto-creation are disabled by default. Production topic policy stays in
  infrastructure-as-code.

The delivery guarantee is **at least once**, not exactly once. Handlers must be idempotent,
or use a database inbox/outbox/transactional design for external side effects.

## Install

```bash
pip install fastkafka2
```

Requirements: Python 3.10 or newer, Pydantic 2, and `confluent-kafka` 2.x.

## Minimal production service

```python
import os

from pydantic import BaseModel, ConfigDict

from fastkafka2 import KafkaApp, KafkaHandler, KafkaMessage, SASL


class Order(BaseModel):
    model_config = ConfigDict(extra="forbid")

    order_id: str
    amount_minor: int


class EventHeaders(BaseModel):
    model_config = ConfigDict(extra="forbid")

    event_type: str
    trace_id: str


orders = KafkaHandler()


@orders("orders", headers_filter={"event_type": "created"})
async def consume_order(message: KafkaMessage[Order, EventHeaders]) -> None:
    # `order_service` is your own application service. Keep the operation idempotent by
    # order_id: at-least-once delivery can redeliver around crashes and rebalances.
    await order_service.apply_once(message.data.order_id, message.data.amount_minor)


security = SASL(
    mechanism="SCRAM-SHA-512",
    username=os.environ["KAFKA_USERNAME"],
    password=os.environ["KAFKA_PASSWORD"],
    ca_location=os.environ["KAFKA_CA_FILE"],
)

app = KafkaApp(
    title="orders-consumer",
    description="Consumes order events",
    bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"],
    group_id="orders-consumer-v1",
    security=security,
    on_error="dlt",
    max_message_bytes=4 * 1024 * 1024,
    max_total_queue_size=10_000,
    max_total_queue_bytes=256 * 1024 * 1024,
    max_concurrent_handlers=128,
)
app.include_handler(orders)
```

Start with `await app.start()` and always pair it with `await app.stop()`, or call
`await app.run()` for built-in SIGINT/SIGTERM handling. The application-owned producer is
available as `app.producer` and follows the same lifecycle.

## TLS and local development

TLS with broker verification:

```python
from fastkafka2 import TLS

security = TLS(ca_location="/run/secrets/kafka-ca.pem")
```

mTLS adds `certificate_location` and `key_location`. Hostname and certificate verification
cannot be disabled.

Local plaintext Kafka must be explicit:

```python
from fastkafka2 import Plaintext

security = Plaintext()  # local Docker/test broker only
```

Do not select `Plaintext()` from an untrusted environment variable. Choose the security
object in deployment code and mount secrets from the platform secret store.

## Authentication before decoding

For Kafka, authentication happens at the transport/session layer:

```text
TLS verification -> SASL authentication -> broker authorizes/fetches records
-> UTF-8 metadata checks -> header routing -> JSON decode -> Pydantic validation -> handler
```

Therefore an unauthenticated client connection cannot deliver records to the consumer for
JSON/Pydantic parsing. Header filters also run before body decoding, but they are routing,
not an authentication mechanism. If message-level signatures are required, verify signed
metadata before accepting a handler and keep broker ACLs enabled.

## Poison-message policies

`on_error` is explicit and fail-closed:

| Policy | Behaviour | Source offset |
|---|---|---|
| `pause` (default) | Pauses the native Kafka partition | Not advanced |
| `dlt` | Awaits DLT broker acknowledgement | Advanced only after DLT success |
| `skip` | Drops the record | Requires `allow_data_loss=True` |

With `dlt`, provision `<source-topic>.dead` before deployment. `auto_create_dlt=True` is
intended for controlled environments; its replication factor defaults to 3. DLT records
contain the raw value and origin metadata. Source key, source headers, and exception text are
excluded unless explicitly enabled because they may contain credentials or personal data.

DLT publication and source-offset commit are not one Kafka transaction. A crash between
those operations can duplicate a DLT record, but does not silently lose the source record.

## High-load controls

Important `KafkaApp`/`KafkaConsumerService` limits:

| Option | Default | Purpose |
|---|---:|---|
| `consume_batch` | 500 | Records requested per consume call |
| `max_queue_size` | 1,000 | Per-partition queued records |
| `max_total_queue_size` | 10,000 | Global queued/in-flight record count |
| `max_total_queue_bytes` | 256 MiB | Global raw-payload byte budget |
| `max_concurrent_handlers` | 256 | Global business-handler concurrency |
| `max_message_bytes` | 4 MiB | Accepted value size and per-message fetch limit |
| `max_header_bytes` | 64 KiB | Accepted header budget |
| `max_key_bytes` | 64 KiB | Accepted key budget |
| `max_fetch_bytes` | 64 MiB | Maximum fetch response budget |
| `commit_interval` | 0.5 s | Periodic synchronous batch commit interval |

Tune handler concurrency from downstream capacity, not CPU count alone. Keep queue byte
budgets below the container memory limit with room for librdkafka, decoded Pydantic objects,
application caches, and temporary batches.

`get_stats()` exposes configured limits, buffered count/bytes, and per-partition processed,
failed, poison, queue, and pause state. Export these values to your metrics system and alert
on sustained queue growth, paused partitions, commit errors in logs, consumer lag, and DLT
rate.

## Producer

```python
producer = app.producer

await producer.send_message(
    "order-results",
    {"order_id": "o-1", "status": "accepted"},
    key="o-1",
)

delivered = await producer.send_many(
    "order-results",
    [{"order_id": f"o-{i}"} for i in range(1_000)],
)
```

The producer enforces idempotence, `acks=all`, a bounded local queue, per-message/header
limits, and batch count/byte limits. `send_many` waits for every enqueued delivery report;
partial enqueue or delivery failure raises.

## Topic provisioning

Create topics outside the application with explicit partitions, replication factor,
retention/compaction policy, min ISR, and ACLs. `auto_create_topics=False` is the default.
For a single-node local broker only:

```python
app = KafkaApp(
    title="Order Service",
    description="Validates and applies order events",
    bootstrap_servers="localhost:9092",
    group_id="order_service",
    security=Plaintext(),
    auto_create_topics=True,
    topic_num_partitions=4,
    topic_replication_factor=1,
)
```

## Validation and routing

Every handler needs `KafkaMessage[DataModel, HeadersModel]`, with a concrete Pydantic data
model; the headers slot may be a plain `dict`. A bare `KafkaMessage` is rejected at
registration. Invalid UTF-8 metadata, an oversized record, invalid JSON, schema failure, or
no matching handler all follow the poison policy without advancing the offset accidentally.

For a handler that does not read the body — a router, an audit sink, a catch-all — pass
`require_data_model=False` on the handler or on its `KafkaHandler` group to receive the body
as decoded and unvalidated. The registration logs a warning naming the handler and topic.

```python
audit = KafkaHandler(require_data_model=False)


@audit("events")
async def sink(message: KafkaMessage) -> None:
    ...
```

Body decoding is lazy: raw bytes are decoded only after metadata validation and a matching
handler are established.

## Release checks

From this directory:

```bash
python -m pytest -q -m "not integration"
python -m ruff check fastkafka2 tests benchmarks
python -m mypy fastkafka2
python -m bandit -q -r fastkafka2
python -m pip_audit
python -m build --outdir release-dist/$VERSION
python -m twine check release-dist/$VERSION/*
```

Build into a fresh, version-specific directory. `dist/` accumulates artifacts from earlier
releases, and `twine check dist/*` — or worse, `twine upload dist/*` — would pick them up.

Integration and throughput tests require the local broker described in
`docker-compose.test.yml`; they refuse the previously known production address.

See [SAFETY.md](SAFETY.md) for the threat model and [CHANGELOG.md](CHANGELOG.md) for the
0.6 migration notes.
