Metadata-Version: 2.5
Name: py-drakkar
Version: 1.17.0
Summary: Kafka subprocess orchestration framework
Author: wlame
License-Expression: MIT
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.13
Requires-Dist: aiosqlite>=0.20.0
Requires-Dist: asyncpg>=0.30.0
Requires-Dist: confluent-kafka>=2.6.0
Requires-Dist: fastapi>=0.115.0
Requires-Dist: httpx>=0.27.0
Requires-Dist: jinja2>=3.1.0
Requires-Dist: prometheus-client>=0.21.0
Requires-Dist: pydantic-settings>=2.14.2
Requires-Dist: pydantic>=2.10.0
Requires-Dist: pymongo>=4.16.0
Requires-Dist: pyyaml>=6.0.2
Requires-Dist: redis>=5.2.0
Requires-Dist: structlog>=24.4.0
Requires-Dist: uvicorn[standard]>=0.32.0
Provides-Extra: dev
Requires-Dist: httpx2>=2.9.1; extra == 'dev'
Requires-Dist: httpx>=0.27.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=1.3.0; extra == 'dev'
Requires-Dist: pytest-cov>=7.1.0; extra == 'dev'
Requires-Dist: pytest>=9.0.3; extra == 'dev'
Requires-Dist: ruff>=0.11.0; extra == 'dev'
Provides-Extra: perf
Requires-Dist: orjson>=3.9; extra == 'perf'
Description-Content-Type: text/markdown

# Drakkar

**Kafka → subprocess pool → sinks, for Python 3.13+.**

Drakkar is an orchestration framework for CPU-heavy stream processing: it consumes messages from Kafka, turns them into invocations of an external binary run in a managed subprocess pool, and delivers the results to any combination of Kafka, PostgreSQL, MongoDB, Redis, HTTP, and files. You write a handler with a few async hooks; the framework owns polling, windowing, backpressure, delivery, offset commits, and observability.

Workers are the Drakkars, executors are the Vikings.

```mermaid
flowchart LR
    K["Kafka<br>source topic"] -- "poll" --> W
    subgraph worker ["Drakkar worker — one pipeline per partition"]
        W["window of<br>messages"] --> A["arrange()<br>your code"]
        A -- "tasks" --> P["subprocess pool<br>runs your binary"]
        P -- "results" --> T["on_task_complete()<br>your code"]
    end
    T -- "payloads" --> S["sinks — any combination<br>Kafka · Postgres · MongoDB · Redis · HTTP · files"]
    S -- "failed delivery" --> DLQ["DLQ topic"]
    S -- "all confirmed" --> CO["commit offsets<br>(watermark)"]
```

<!-- TODO(wlame): screenshot or GIF of the Live executor timeline goes here.
<p align="center"><img src="docs/media/timeline.gif" alt="Live executor timeline" width="800"></p>
-->

**[Documentation](https://wlame.github.io/drakkar)** — full guides for every feature below.

## Features

- **Per-partition pipelines** with watermark offset tracking — commits happen only after every sink confirmed
- **Pluggable sinks** — Kafka, PostgreSQL, MongoDB, Redis, HTTP, filesystem; multiple named instances per type, third-party sinks via entry points
- **Dead letter queue** with replay tooling; `on_delivery_error()` decides retry / skip / DLQ per failure
- **Backpressure** via Kafka pause/resume — memory stays bounded regardless of lag
- **Typed messages** — Pydantic models as type parameters, auto de/serialization
- **Operator UI** — live executor timeline, partition lag, message tracing, and a Message Probe that runs a pasted message through the full pipeline with zero footprint on production state
- **UI customization** — handler-defined probe tabs, links/badges/formats on any field, declared dashboard pages — all server-side, no client code
- **Cache** (optional) — `self.cache` key/value store with write-behind SQLite and peer sync across workers
- **Offload** — `await self.offload(fn, ...)` keeps CPU-bound hook work off the event loop
- **Webapp** (optional) — the same handler pipeline exposed as a synchronous HTTP endpoint with auth and rate limits
- **Observability** — Prometheus metrics, ECS-compatible structured logging, flight recorder (SQLite event log), runtime-health and host-pressure monitors, task cost/throughput stats
- **Kubernetes-ready** — `/healthz` and `/readyz` probes, reference manifests, crash/OOM detection on restart

A **Go implementation** ([drakkar-go](https://github.com/wlame/drakkar-go)) is config- and contract-compatible; both backends serve the same [drakkar-ui](https://github.com/wlame/drakkar-ui) web UI and can run side by side in one fleet.

## Quick start

```bash
uv init my-processor && cd my-processor
uv add py-drakkar
```

```python
# handler.py
from pydantic import BaseModel
from drakkar import (
    BaseDrakkarHandler, CollectResult, ExecutorTask,
    KafkaPayload, PostgresPayload, make_task_id,
)

class JobInput(BaseModel):
    job_id: str
    command: str

class JobOutput(BaseModel):
    job_id: str
    result: str

class MyHandler(BaseDrakkarHandler[JobInput, JobOutput]):
    async def arrange(self, messages, pending):
        # window of Kafka messages -> subprocess tasks
        return [
            ExecutorTask(
                task_id=make_task_id('job'),
                args=['--cmd', msg.payload.command],
                source_offsets=[msg.offset],
                metadata={'job_id': msg.payload.job_id},
            )
            for msg in messages
        ]

    async def on_task_complete(self, result):
        # subprocess output -> sink payloads
        output = JobOutput(
            job_id=result.task.metadata['job_id'],
            result=result.stdout.strip(),
        )
        return CollectResult(
            kafka=[KafkaPayload(data=output, key=output.job_id.encode())],
            postgres=[PostgresPayload(table='results', data=output)],
        )
```

```yaml
# drakkar.yaml
kafka:
  brokers: "localhost:9092"
  source_topic: "jobs"
  consumer_group: "my-workers"

executor:
  binary_path: "/usr/local/bin/my-tool"
  max_executors: 8
  task_timeout_seconds: 60

sinks:
  kafka:
    job_results_out:
      topic: "job-results"
  postgres:
    main_db:
      dsn: "postgresql://user:pass@localhost:5432/mydb"
```

```python
# main.py
from drakkar import DrakkarApp
from handler import MyHandler

DrakkarApp(handler=MyHandler(), config_path='drakkar.yaml').run()
```

```bash
WORKER_ID=worker-1 python main.py
```

Every config field can be overridden by environment variables (`DK_` prefix, `__` for nesting: `DK_EXECUTOR__MAX_EXECUTORS=16`). Scale horizontally by running more workers in the same consumer group; cooperative-sticky rebalancing spreads partitions without stopping the others.

More hooks are available for aggregation and error handling — `on_message_complete` (fan-out → fan-in), `on_window_complete`, `on_error`, `@periodic` background tasks, lifecycle hooks. See the [handler guide](https://wlame.github.io/drakkar/handler/).

## Try it

A full docker-compose environment with Kafka, all six sink types, five workers, and a load generator lives in [`integration/`](integration/):

```bash
cd integration
docker compose up --build
```

Then open `http://localhost:8081` for the first worker's UI. See the [integration guide](https://wlame.github.io/drakkar/integration/).

## Development

[`just`](https://github.com/casey/just) is the dev entrypoint; CI runs the same recipes.

```bash
just install     # uv sync with dev + perf extras
just test        # unit tests (hermetic, no network)
just ci          # format check -> lint -> types -> tests + coverage gate
just docs-serve  # live-reload docs at http://127.0.0.1:8000
just --list      # everything else (integration env, chaos test, DLQ replay, ...)
```

See [`docs/development.md`](docs/development.md) and [`CONTRIBUTING.md`](CONTRIBUTING.md).

## License

MIT
