Metadata-Version: 2.4
Name: easy-faststream
Version: 0.10.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: prometheus-client<1,>=0.21; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: python-telegram-bot<23,>=22; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Requires-Dist: twine>=6.0; extra == 'dev'
Provides-Extra: observability
Requires-Dist: prometheus-client<1,>=0.21; extra == 'observability'
Provides-Extra: telegram
Requires-Dist: python-telegram-bot<23,>=22; extra == 'telegram'
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/*
```

## Persistent retry audit

Install ClickHouse support:

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

Configure an audit sink:

```python
from easy_faststream import ClickHouseAuditSink, StreamApp

audit_sink = ClickHouseAuditSink(
    table="bronze.easy_faststream_retry_audit",
)

stream = StreamApp(
    audit_sink=audit_sink,
)
```

The framework records these lifecycle statuses:

- `received`
- `validation_failed`
- `processing_failed`
- `retry_scheduled`
- `retry_succeeded`
- `non_retryable_failed`
- `dead_lettered`
- `succeeded`

Audit writing is best-effort. If the audit database is temporarily unavailable,
the failure is logged without interrupting business-message processing.


## Durable audit delivery

For production environments, audit records can first be published to a durable
RabbitMQ queue. A separate worker then writes them to ClickHouse.

### Business application

```python
from easy_faststream import RabbitMQAuditSink, StreamApp

stream = StreamApp.from_env()

stream.audit_sink = RabbitMQAuditSink(
    stream.broker,
    exchange="ex.easy_faststream.audit",
    queue="p_q.easy_faststream.audit",
    routing_key="easy.audit",
)
```

### Audit worker

```python
from easy_faststream import (
    ClickHouseAuditSink,
    StreamApp,
    register_clickhouse_audit_worker,
)

stream = StreamApp.from_env()

sink = ClickHouseAuditSink(
    table="bronze.easy_faststream_retry_audit",
)

register_clickhouse_audit_worker(
    stream,
    sink,
    queue="p_q.easy_faststream.audit",
    exchange="ex.easy_faststream.audit",
    routing_key="easy.audit",
    retries=5,
    retry_delay=5,
    retry_backoff=2,
)

app = stream.app
```

The business consumer declares the durable audit exchange, queue, and binding.
Audit events therefore remain queued even when the ClickHouse audit worker is
offline.

The worker uses durable retries and sends exhausted failures to its own DLQ.
The audit worker must not configure another audit sink, preventing recursive
audit messages.

## Schema-first publishing

Create a reusable typed publisher from `StreamApp`:

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

from pydantic import BaseModel

from easy_faststream import StreamApp


class TripRequested(BaseModel):
    order_id: UUID
    service_type: str
    estimated_fare: float
    event_at: datetime


stream = StreamApp.from_env()

publisher = stream.publisher(
    event="passapp.trip.requested",
    schema=TripRequested,
    exchange="ex.passapp.event.trip",
    schema_version="1",
)
```

Publish a validated event:

```python
async with stream.broker:
    result = await publisher.publish(
        {
            "order_id": "051fe640-7650-4817-afb1-091b527d4d81",
            "service_type": "RICKSHAW",
            "estimated_fare": 5000,
            "event_at": datetime.now(),
        },
        headers={
            "x-source-service": "booking-api",
        },
    )

print(result.message_id)
print(result.correlation_id)
```

The publisher automatically:

- Validates outgoing payloads with Pydantic.
- Converts UUID and datetime values to JSON-compatible values.
- Declares a durable topic exchange.
- Publishes persistent RabbitMQ messages.
- Generates a message ID when one is not provided.
- Uses the message ID as the default correlation ID.
- Adds event name, schema name, and schema version headers.

Invalid payloads raise Pydantic `ValidationError` before anything is published.

## Consumer idempotency

Consumers can skip duplicate messages using an idempotency store:

```python
from easy_faststream import (
    InMemoryIdempotencyStore,
    StreamApp,
)

stream = StreamApp.from_env()
store = InMemoryIdempotencyStore()


@stream.consumer(
    event="passapp.trip.completed",
    schema=TripCompleted,
    queue="p_q.passapp.trip.completed",
    exchange="ex.passapp.event.trip",
    idempotency_store=store,
    idempotency_key="order_id",
    idempotency_lock_ttl=300,
    idempotency_retention=86400,
)
async def consume(event: TripCompleted) -> None:
    ...
```

If `idempotency_key` is omitted, the RabbitMQ message ID is used:

```python
@stream.consumer(
    event="passapp.trip.completed",
    schema=TripCompleted,
    idempotency_store=store,
)
async def consume(event: TripCompleted) -> None:
    ...
```

A callable key is also supported:

```python
idempotency_key=lambda event: (
    f"{event.order_id}:{event.event_at}"
)
```

The framework:

- Acquires a processing lease after schema validation.
- Prevents concurrent processing of the same key.
- Releases the lease when processing fails, allowing retries.
- Marks the key completed after the handler and sink succeed.
- ACKs completed duplicates without executing the handler.
- Records `duplicate_skipped` in the audit history.
- Uses ownership tokens so expired workers cannot modify newer leases.

### In-memory store limitation

`InMemoryIdempotencyStore` is intended for tests, development, or one consumer
process. Its state is not shared between processes and is lost when the
application restarts.

For RabbitMQ-to-ClickHouse pipelines running multiple consumers, configure
`ClickHouseSink` with a stable business idempotency key:

```python
sink = ClickHouseSink(
    table="bronze.trip_completed",
    idempotency_key="order_id",
)
```

ClickHouse sink deduplication protects the final table but does not prevent the
consumer handler itself from running more than once. RabbitMQ should still be
treated as an at-least-once delivery system.

## Telegram monitoring bot

Install Telegram and ClickHouse support:

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

The monitoring bot queries the ClickHouse audit table and provides near-real-time
statistics through Telegram commands and inline buttons.

Available views include:

- Last 5 minutes
- Last 1 hour
- Last 24 hours
- Success rate
- Validation and processing failures
- Scheduled and successful retries
- Non-retryable failures
- Dead-lettered messages
- Duplicates skipped
- Recent error details

### Environment variables

```env
EASY_STREAM_TELEGRAM_BOT_TOKEN=your-bot-token
EASY_STREAM_TELEGRAM_ALLOWED_CHAT_IDS=-1001234567890
EASY_STREAM_TELEGRAM_AUDIT_TABLE=bronze.easy_faststream_retry_audit
```

Never commit or log the Telegram bot token. Restrict access with an explicit
chat-ID allowlist.

### Example

```python
from easy_faststream import (
    ClickHouseMonitoringRepository,
    TelegramMonitoringBot,
)

repository = ClickHouseMonitoringRepository(
    table="bronze.easy_faststream_retry_audit",
)

bot = TelegramMonitoringBot(
    token="loaded-from-environment",
    repository=repository,
    allowed_chat_ids={-1001234567890},
)

bot.run_polling()
```

Send `/start` or `/status` to the bot. The bot uses inline buttons to refresh
statistics and view recent errors.

Statistics are generated from persisted audit records and are near-real-time.
They do not currently indicate whether an individual consumer process is online.
Consumer heartbeats can be added in a future release.


Observability
Install Prometheus support:
```bash
pip install "easy-faststream[observability]"
```
Enable metrics and health endpoints through environment variables:
```env
EASY_STREAM_METRICS_ENABLED=true
EASY_STREAM_METRICS_HOST=0.0.0.0
EASY_STREAM_METRICS_PORT=8000

EASY_STREAM_HEALTH_ENABLED=true
EASY_STREAM_HEALTH_HOST=0.0.0.0
EASY_STREAM_HEALTH_PORT=8080
```
No additional application setup is required:
```python
from easy_faststream import StreamApp

stream = StreamApp.from_env()
app = stream.app
```
Available endpoints:
`GET :8000/metrics` — Prometheus metrics
`GET :8080/health` — process liveness
`GET :8080/ready` — application readiness
The framework exposes metrics for:
Message lifecycle statuses
Processing duration
Scheduled retries
Buffered-sink queue depth
ClickHouse batch sizes
Batch write duration
Batch failures
Example Prometheus configuration:
```yaml
scrape_configs:
  - job_name: easy-faststream
    scrape_interval: 15s
    static_configs:
      - targets:
          - consumer-host:8000
```
For a buffered ClickHouse sink, reuse the application's collector:
```python
from easy_faststream import BufferedClickHouseSink

sink = BufferedClickHouseSink(
    table="bronze.events",
    batch_size=1000,
    flush_interval=2,
    metrics=stream.metrics,
    metrics_name="bronze-events",
)
```