Metadata-Version: 2.4
Name: easy-faststream
Version: 0.3.0
Summary: A schema-first FastStream framework with durable retries, dead-letter handling, and optional ClickHouse support.
Author: Ek
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.11
Requires-Dist: faststream[cli,rabbit]<0.8,>=0.7.1
Requires-Dist: pydantic-settings<3,>=2.2
Requires-Dist: pydantic<3,>=2.7
Provides-Extra: clickhouse
Requires-Dist: clickhouse-connect<1,>=0.8; extra == 'clickhouse'
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == 'dev'
Requires-Dist: clickhouse-connect<1,>=0.8; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Requires-Dist: twine>=6.0; extra == 'dev'
Description-Content-Type: text/markdown

# easy-faststream

`easy-faststream` is a schema-first wrapper around FastStream for RabbitMQ.

Application developers define a Pydantic schema and consumer function. The
framework handles validation, acknowledgements, durable retries, dead-letter
routing, and optional idempotent ClickHouse insertion.

## Installation

Install the RabbitMQ framework:

```bash
pip install easy-faststream
```

Install it with ClickHouse support:

```bash
pip install "easy-faststream[clickhouse]"
```

## Basic consumer

```python
from datetime import datetime
from uuid import UUID

from pydantic import BaseModel

from easy_faststream import MessageContext, StreamApp


class TripRequested(BaseModel):
    order_id: UUID
    passenger_id: UUID | None = None
    service_type: str
    event_at: datetime


stream = StreamApp.from_env()


@stream.consumer(
    event="passapp.trip.requested",
    schema=TripRequested,
    queue="p_q.passapp.trip.requested",
    exchange="ex.passapp.event.trip",
    routing_key="passapp.trip.requested",
)
async def consume_trip(
    event: TripRequested,
    context: MessageContext,
) -> None:
    print(event.order_id, context.message_id)


app = stream.app
```

Run the application:

```bash
python -m faststream run app:app
```

## RabbitMQ configuration

```env
EASY_STREAM_RABBITMQ_URL=amqp://guest:guest@localhost:5672/
EASY_STREAM_APP_NAME=trip-consumer

EASY_STREAM_DEFAULT_EXCHANGE=easy.events
EASY_STREAM_RETRY_EXCHANGE=easy.events.retry
EASY_STREAM_DLQ_EXCHANGE=easy.events.dead

EASY_STREAM_RETRY_QUEUE_SUFFIX=.retry
EASY_STREAM_DLQ_QUEUE_SUFFIX=.dead

EASY_STREAM_MAX_RETRIES=3
EASY_STREAM_RETRY_DELAY_SECONDS=1
EASY_STREAM_RETRY_BACKOFF=2

EASY_STREAM_GRACEFUL_TIMEOUT=30
```

With the settings above, retry delays are 1, 2, and 4 seconds.

Retries use durable RabbitMQ TTL queues. Retry messages therefore survive
consumer shutdowns and application restarts.

## ClickHouse sink

Configure the ClickHouse connection:

```env
EASY_STREAM_CLICKHOUSE_HOST=localhost
EASY_STREAM_CLICKHOUSE_PORT=8123
EASY_STREAM_CLICKHOUSE_USERNAME=default
EASY_STREAM_CLICKHOUSE_PASSWORD=
EASY_STREAM_CLICKHOUSE_DATABASE=default
EASY_STREAM_CLICKHOUSE_SECURE=false
```

Attach a sink to a consumer:

```python
from easy_faststream import ClickHouseSink, StreamApp


stream = StreamApp.from_env()

sink = ClickHouseSink(
    table="bronze.trip_requested",
    idempotency_key="order_id",
)


@stream.consumer(
    event="passapp.trip.requested",
    schema=TripRequested,
    queue="p_q.passapp.trip.requested",
    exchange="ex.passapp.event.trip",
    routing_key="passapp.trip.requested",
    sink=sink,
)
async def consume_trip(event: TripRequested) -> None:
    print(f"Processing order {event.order_id}")


app = stream.app
```

The processing order is:

```text
Validate schema
    → Run handler
    → Insert into ClickHouse
    → ACK RabbitMQ message
```

If ClickHouse insertion fails, the RabbitMQ message is not considered
successfully processed.

## ClickHouse idempotency

The sink creates a stable SHA-256 `insert_deduplication_token` using:

- ClickHouse table
- RabbitMQ routing key
- Configured business key, such as `order_id`

Publishing the same business event with different RabbitMQ message IDs therefore
uses the same ClickHouse token.

The destination table must use a `MergeTree` family engine with an appropriate
deduplication window. ClickHouse deduplication is limited by that window and is
not a permanent unique-key constraint.

## Permanent and transient failures

Transient failures, such as connection timeouts, enter the durable retry flow.

Permanent ClickHouse errors are sent directly to the DLQ without unnecessary
retries. Examples include:

- Missing table or database
- Unknown column
- Type mismatch
- Invalid input
- SQL syntax error

## Failure behavior

- Invalid schema: publish the original payload and validation error to the DLQ.
- Processing error: use durable retry queues, then publish to the DLQ.
- Transient ClickHouse error: use the durable retry flow.
- Permanent ClickHouse error: publish directly to the DLQ.
- `NonRetryableError`: skip retries and publish directly to the DLQ.
- Retry publication failure: NACK and requeue the original message.
- DLQ publication failure: NACK and requeue the original message.
- Successful processing: ACK the original message.
- Successful DLQ publication: ACK the original message.

## Development

Install development dependencies:

```bash
python -m pip install -e ".[dev]"
```

Run checks:

```bash
python -m ruff check src tests examples
python -m pytest
python -m build
python -m twine check dist/*
```