Metadata-Version: 2.4
Name: easy-faststream
Version: 0.11.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: opentelemetry-api<2,>=1.30; extra == 'dev'
Requires-Dist: opentelemetry-exporter-otlp-proto-http<2,>=1.30; extra == 'dev'
Requires-Dist: opentelemetry-sdk<2,>=1.30; 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'
Provides-Extra: tracing
Requires-Dist: opentelemetry-api<2,>=1.30; extra == 'tracing'
Requires-Dist: opentelemetry-exporter-otlp-proto-http<2,>=1.30; extra == 'tracing'
Requires-Dist: opentelemetry-sdk<2,>=1.30; extra == 'tracing'
Description-Content-Type: text/markdown

easy-faststream
`easy-faststream` is a schema-first framework built on FastStream for reliable
RabbitMQ data pipelines.
Application developers define Pydantic schemas and consumer functions. The
framework handles validation, acknowledgements, durable retries, dead-letter
routing, typed publishing, idempotency, ClickHouse delivery, audit history,
metrics, health checks, distributed tracing, structured logging, and Telegram
monitoring.
Features
Schema-first RabbitMQ consumers and publishers using Pydantic.
Durable exponential-backoff retries using RabbitMQ TTL queues.
Dead-letter routing with the original payload and failure details.
Retry-safe acknowledgements and requeue behavior.
Transient and non-retryable failure classification.
ClickHouse sinks with stable deduplication tokens.
Buffered ClickHouse batch insertion with backpressure.
Persistent or durable retry audit delivery.
Consumer idempotency without requiring Redis.
Prometheus metrics and alerting rules.
Liveness and readiness HTTP endpoints.
OpenTelemetry tracing with W3C context propagation.
Structured JSON logging with sensitive-field redaction.
ClickHouse and live-runtime monitoring through Telegram.
Installation
Install the RabbitMQ framework:
```bash
pip install easy-faststream
```
Install it with ClickHouse support:
```bash
pip install "easy-faststream[clickhouse]"
```
Install observability, tracing, and Telegram support:
```bash
pip install \
  "easy-faststream[clickhouse,observability,tracing,telegram]"
```
Python 3.11 or newer is required.
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
EASY_STREAM_RABBITMQ_PREFETCH_COUNT=100
```
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.
`EASY_STREAM_RABBITMQ_PREFETCH_COUNT` limits the number of unacknowledged
messages delivered to each consumer channel. Tune it together with sink batch
size, message-processing time, memory, and consumer concurrency.
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.
Buffered ClickHouse sink
Use `BufferedClickHouseSink` for higher-throughput pipelines. It groups rows
into ClickHouse inserts and flushes when either `batch_size` is reached or
`flush_interval` expires.
```python
from easy_faststream import (
    BufferedClickHouseSink,
    StreamApp,
)


stream = StreamApp.from_env()

sink = BufferedClickHouseSink(
    table="bronze.trip_requested",
    batch_size=1000,
    flush_interval=2.0,
    max_buffer_size=10000,
    idempotency_key="order_id",
    metrics=stream.metrics,
    metrics_name="trip-requested",
)


@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(event.order_id)


app = stream.app
```
When attached to a consumer, the sink is managed by `StreamApp` and is closed
gracefully during shutdown. Each consumer call completes only after its row has
been inserted successfully. A failed batch propagates the error back to every
message in that batch, allowing the framework retry policy to handle it.
The bounded buffer applies backpressure when `max_buffer_size` is reached. For
best throughput, configure RabbitMQ prefetch high enough to fill batches without
allowing excessive unacknowledged messages.
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`
`duplicate_skipped`
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.
Creates a producer trace span and injects W3C trace context when tracing is
enabled.
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,
    RuntimeMonitoringClient,
    TelegramMonitoringBot,
)

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

runtime = RuntimeMonitoringClient(
    health_url="http://consumer-host:8080/ready",
    metrics_url="http://consumer-host:8000/metrics",
)

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

bot.run_polling()
```
Available commands:
`/start` — display the monitoring menu.
`/status` — show audit statistics for the default time window.
`/health` — show application readiness and managed-component status.
`/buffers` — show live buffered ClickHouse sink statistics.
The same actions are available as inline buttons. `/status` and recent failures
are generated from persisted audit records. `/health` and `/buffers` query the
live HTTP endpoints configured in `RuntimeMonitoringClient`.
The Telegram process must be able to reach the consumer's health and metrics
ports. Keep these endpoints on a private network or protect them with your
platform's network access controls.

Observability
Install Prometheus support:
```bash
pip install "easy-faststream[observability]"
```
Enable metrics and health endpoints:
```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:
Endpoint	Purpose	Success response
`GET :8000/metrics`	Prometheus metrics	Prometheus text format
`GET :8080/health`	Process liveness	`{"status": "alive"}`
`GET :8080/ready`	Application readiness	HTTP 200 when ready
The framework exposes metrics for:
Message lifecycle statuses.
Message-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
```
The health and metrics servers are managed components. They start and stop with
the FastStream application. The readiness response includes the application
phase, start time, managed-component count, started-component count, and the
startup error type when startup fails.
Prometheus alerting
The project includes an `easy-faststream-alerts.yml` rules file. Configure
Prometheus to load it:
```yaml
rule_files:
  - /etc/prometheus/rules/easy-faststream-alerts.yml
```
The rules cover:
Missing `easy-faststream` targets.
Dead-lettered messages.
ClickHouse batch failures.
High processing-failure rate.
High p95 message-processing latency.
A non-empty ClickHouse buffer that remains stuck.
High retry volume.
Validate the rule file before deployment:
```bash
promtool check rules easy-faststream-alerts.yml
```
Alert thresholds are starting points. Adjust them for your normal message
volume, latency, batch size, and operational response policy.
OpenTelemetry tracing
Install tracing support:
```bash
pip install "easy-faststream[tracing]"
```
Enable OTLP trace export:
```env
EASY_STREAM_TRACING_ENABLED=true
EASY_STREAM_TRACING_SERVICE_NAME=trip-consumer
EASY_STREAM_TRACING_OTLP_ENDPOINT=http://otel-collector:4318/v1/traces
EASY_STREAM_TRACING_SAMPLE_RATIO=1.0
EASY_STREAM_TRACING_TIMEOUT_SECONDS=10
```
`StreamApp.from_env()` automatically creates and manages the OpenTelemetry
provider when tracing is enabled.
The framework:
Creates producer spans around schema-first publishing.
Creates consumer spans around message processing.
Injects W3C `traceparent` context into RabbitMQ headers.
Extracts parent context in consumers.
Preserves trace context through durable retries and DLQ delivery.
Adds messaging, event, schema, message, and correlation attributes.
Shuts down its owned trace provider during graceful application shutdown.
The OTLP endpoint must point to an OpenTelemetry-compatible collector or
backend, such as the OpenTelemetry Collector, Grafana Tempo, or Jaeger with
OTLP HTTP enabled. If you do not run an OTLP receiver yet, keep tracing
disabled; metrics, health checks, logging, and Telegram monitoring continue to
work independently.
For production, use an appropriate sample ratio. For example, `0.1` samples
approximately 10% of new traces while retaining parent-based sampling
decisions.
Structured JSON logging
Enable framework JSON logs:
```env
EASY_STREAM_JSON_LOGGING_ENABLED=true
EASY_STREAM_LOG_LEVEL=INFO
```
Supported levels are `DEBUG`, `INFO`, `WARNING`, `ERROR`, and `CRITICAL`.
The JSON formatter includes:
UTC timestamp.
Log level and logger name.
Event message and structured fields.
Exception information.
Active OpenTelemetry trace and span IDs when available.
Redaction for password, token, secret, authorization, cookie, and API-key
fields.
Example:
```json
{
  "timestamp": "2026-07-28T08:55:50.396197+00:00",
  "level": "INFO",
  "logger": "easy_faststream",
  "message": "message_processed",
  "event": "passapp.trip.requested",
  "message_id": "message-001",
  "retry_count": 0
}
```
The setting configures the `easy_faststream` logger. FastStream's own logger may
continue using its configured console format unless the application configures
it separately.
Complete environment example
```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_MAX_RETRIES=3
EASY_STREAM_RETRY_DELAY_SECONDS=1
EASY_STREAM_RETRY_BACKOFF=2
EASY_STREAM_GRACEFUL_TIMEOUT=30
EASY_STREAM_RABBITMQ_PREFETCH_COUNT=100

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

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

EASY_STREAM_TRACING_ENABLED=false
EASY_STREAM_TRACING_SERVICE_NAME=trip-consumer
EASY_STREAM_TRACING_OTLP_ENDPOINT=http://localhost:4318/v1/traces
EASY_STREAM_TRACING_SAMPLE_RATIO=1.0
EASY_STREAM_TRACING_TIMEOUT_SECONDS=10

EASY_STREAM_JSON_LOGGING_ENABLED=true
EASY_STREAM_LOG_LEVEL=INFO
```
Percent-encode reserved characters in RabbitMQ usernames, passwords, and
virtual-host names before placing them in an AMQP URL. Never commit `.env`
files or credentials.