Metadata-Version: 2.4
Name: easy-faststream
Version: 0.12.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 event processing.
Application developers define Pydantic schemas and consumer functions. The
framework handles:
Payload validation
RabbitMQ topology
Durable delayed retries
Dead-letter routing
Non-retryable failures
Schema-aware publishing
Idempotent consumer execution
ClickHouse sinks and buffered batch inserts
Persistent processing audit records
Prometheus metrics and health endpoints
OpenTelemetry tracing and structured JSON logging
Telegram monitoring and automatic alerts
Operational CLI commands
Graceful buffered-sink drain reporting
Requirements
Python 3.11 or newer
RabbitMQ
ClickHouse when using ClickHouse-backed features
Installation
Install the core package:
```bash
python -m pip install easy-faststream
```
Install ClickHouse support:
```bash
python -m pip install "easy-faststream[clickhouse]"
```
Install observability support:
```bash
python -m pip install "easy-faststream[observability]"
```
For local development:
```bash
python -m pip install -e ".[dev]"
```
Quick start
Create `app.py`:
```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)
    print(context.message_id)


app = stream.app
```
Run it:
```bash
python -m faststream run app:app
```
Configuration
`StreamApp.from_env()` loads variables with the `EASY_STREAM_` prefix and reads
`.env` when present.
```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_RABBITMQ_PREFETCH_COUNT=100
EASY_STREAM_GRACEFUL_TIMEOUT=30

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
```
Do not commit `.env` files containing RabbitMQ, ClickHouse, Telegram, or tracing
credentials.
Consumer retries and dead letters
Configure retries globally:
```env
EASY_STREAM_MAX_RETRIES=3
EASY_STREAM_RETRY_DELAY_SECONDS=1
EASY_STREAM_RETRY_BACKOFF=2
```
With these values, retry delays are 1, 2, and 4 seconds. Retry messages are
stored in durable RabbitMQ queues and survive application restarts.
Override retries for one consumer:
```python
@stream.consumer(
    event="passapp.trip.requested",
    schema=TripRequested,
    queue="p_q.passapp.trip.requested",
    exchange="ex.passapp.event.trip",
    retries=5,
)
async def consume_trip(event: TripRequested) -> None:
    ...
```
Raise `NonRetryableError` when retrying cannot resolve a failure:
```python
from easy_faststream import NonRetryableError


async def consume_trip(event: TripRequested) -> None:
    if event.service_type == "UNSUPPORTED":
        raise NonRetryableError("Unsupported service type")
```
Failure behavior:
Invalid schema: publish the original payload and validation details to the
DLQ.
Processing failure: retry according to the retry policy, then publish to the
DLQ.
`NonRetryableError`: skip retries and publish directly to the DLQ.
DLQ publish failure: NACK and requeue the original message.
Successful processing or successful DLQ publishing: ACK the original
message.
The DLQ payload includes the original payload, failure classification, retry
count, retry history, source metadata, and timestamps.
Schema-aware publisher
Create a publisher from `StreamApp`:
```python
publisher = stream.publisher(
    event="passapp.trip.requested",
    schema=TripRequested,
    exchange="ex.passapp.event.trip",
    routing_key="passapp.trip.requested",
    schema_version="1",
)

result = await publisher.publish(
    {
        "order_id": "051fe640-7650-4817-afb1-091b527d4d81",
        "service_type": "RICKSHAW",
        "event_at": "2026-07-29T10:00:00+07:00",
    },
    headers={"x-source-service": "booking-api"},
)

print(result.message_id)
print(result.correlation_id)
print(result.payload)
```
The publisher validates the payload before publishing and supplies standard
event, schema, and schema-version headers.
Idempotent consumers
The package includes an in-memory idempotency store:
```python
from easy_faststream import InMemoryIdempotencyStore, StreamApp


store = InMemoryIdempotencyStore()
stream = StreamApp.from_env()
```
The in-memory implementation requires no Redis. It is intended for a single
process and does not coordinate leases across multiple application instances.
Use a shared durable implementation of the `IdempotencyStore` protocol when
cross-process coordination is required.
ClickHouse sink
Install the ClickHouse extra and configure the 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


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",
    sink=sink,
)
async def consume_trip(event: TripRequested) -> None:
    print("Validated:", event.order_id)
```
The handler completes first. The validated model is then written to
ClickHouse. The sink uses ClickHouse insertion deduplication tokens.
Permanent ClickHouse errors such as an unknown table, unknown database, type
mismatch, or invalid input are raised as
`NonRetryableClickHouseSinkError`. Connection and temporary server errors
remain retryable.
Buffered ClickHouse sink
Use buffered inserts for higher throughput:
```python
from easy_faststream import BufferedClickHouseSink


sink = BufferedClickHouseSink(
    table="bronze.trip_requested",
    sink_name="trip-requested",
    idempotency_key="order_id",
    batch_size=1000,
    flush_interval=2,
    max_buffer_size=10000,
    drain_timeout=30,
)

stream.manage(sink)
```
Rows are flushed when `batch_size` is reached or `flush_interval` expires.
`max_buffer_size` provides backpressure. RabbitMQ prefetch should be sized with
the application concurrency and buffer capacity in mind.
Graceful drain reporting
During shutdown, the sink stops accepting new writes and waits for queued and
in-progress writes. It does not cancel an active ClickHouse insert when the
configured drain timeout is exceeded.
When closing the sink directly:
```python
from easy_faststream import DrainReport, DrainStatus


report: DrainReport = await sink.close()

print(report.status)
print(report.initial_buffer_size)
print(report.drained_rows)
print(report.failed_rows)
print(report.final_buffer_size)
print(report.duration_seconds)

if report.status is DrainStatus.FAILED:
    print("Some rows could not be written")
```
When registered with `stream.manage(sink)`, shutdown is automatic. Drain
results are available through structured logs and Prometheus metrics.
Processing audit
Audit records capture message lifecycle events such as:
`received`
`validation_failed`
`processing_failed`
`retry_scheduled`
`retry_succeeded`
`non_retryable_failed`
`dead_lettered`
`duplicate_skipped`
ClickHouse audit sink
```python
from easy_faststream import ClickHouseAuditSink, StreamApp


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

stream = StreamApp.from_env(
    audit_sink=audit_sink,
)
```
Durable RabbitMQ audit path
For a durable audit pipeline, publish audit messages to RabbitMQ and run the
ClickHouse audit worker separately:
```python
from easy_faststream import RabbitMQAuditSink, StreamApp


stream = StreamApp.from_env()

audit_sink = RabbitMQAuditSink(
    stream.broker,
    exchange="ex.easy.audit",
    routing_key="easy.audit",
)
```
This decouples message processing from ClickHouse audit availability.
Metrics
Enable Prometheus metrics:
```env
EASY_STREAM_METRICS_ENABLED=true
EASY_STREAM_METRICS_HOST=0.0.0.0
EASY_STREAM_METRICS_PORT=8000
```
Read metrics:
```bash
curl http://127.0.0.1:8000/metrics
```
Metrics include:
Message lifecycle totals
Processing-duration histograms
Scheduled retries
Buffered row count
Batch size and duration
Batch failures
Drain starts, timeouts, results, rows, and duration
Health and readiness
Enable the health server:
```env
EASY_STREAM_HEALTH_ENABLED=true
EASY_STREAM_HEALTH_HOST=0.0.0.0
EASY_STREAM_HEALTH_PORT=8080
```
Liveness:
```bash
curl http://127.0.0.1:8080/health
```
Readiness:
```bash
curl http://127.0.0.1:8080/ready
```
Readiness reports the application phase and managed-component startup state.
Structured logging
Enable JSON logging:
```env
EASY_STREAM_JSON_LOGGING_ENABLED=true
EASY_STREAM_LOG_LEVEL=INFO
```
Framework events include message processing, retry scheduling, dead-letter
routing, topology declaration, managed-component lifecycle, and buffered-sink
draining. Sensitive values are redacted by the structured logging utilities.
OpenTelemetry tracing
Enable tracing:
```env
EASY_STREAM_TRACING_ENABLED=true
EASY_STREAM_TRACING_SERVICE_NAME=trip-consumer
EASY_STREAM_TRACING_OTLP_ENDPOINT=http://localhost:4318/v1/traces
EASY_STREAM_TRACING_SAMPLE_RATIO=1.0
```
The OTLP endpoint should point to an OpenTelemetry Collector or another
OTLP-compatible tracing backend.
Telegram monitoring
Create a runtime monitoring bot using the ClickHouse audit table and the health
and metrics endpoints:
```python
from easy_faststream import (
    ClickHouseMonitoringRepository,
    RuntimeMonitoringClient,
    TelegramMonitoringBot,
)


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

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

bot = TelegramMonitoringBot(
    token="load-from-environment",
    repository=repository,
    runtime_client=runtime,
)
```
Keep the Telegram token in an environment variable. Never commit it to source
control.
Automatic Telegram alerts
```python
from easy_faststream import (
    TelegramAlertSettings,
    TelegramAlertWorker,
)


alerts = TelegramAlertWorker(
    token="load-from-environment",
    chat_ids={123456789},
    repository=repository,
    runtime_client=runtime,
    settings=TelegramAlertSettings(
        interval_seconds=60,
        cooldown_seconds=300,
        window_minutes=5,
        processing_failure_threshold=5,
        dead_letter_threshold=1,
        buffer_size_threshold=1000,
        batch_failure_threshold=1,
        notify_recovery=True,
    ),
)

stream.manage(alerts)
```
Alerts support failure thresholds, dead letters, runtime readiness, endpoint
availability, high buffer size, batch failures, cooldowns, and recovery
notifications.
Managed components
Objects with asynchronous `start()` and `close()` methods can participate in
the application lifecycle:
```python
stream.manage(component)
```
Managed components start after the FastStream application starts and close
during graceful shutdown.
Operational CLI
Installing the package provides the `easy-faststream` command.
Validate configuration
```bash
easy-faststream config check
```
Test RabbitMQ connectivity:
```bash
easy-faststream config check --connect
```
Return JSON:
```bash
easy-faststream config check --json
```
Passwords are never included in the output.
Check queue status
```bash
easy-faststream queue status \
  --queue p_q.easy_faststream.test \
  --queue p_q.easy_faststream.test.dead
```
Require active consumers:
```bash
easy-faststream queue status \
  --queue p_q.easy_faststream.test \
  --require-consumers
```
Watch every two seconds:
```bash
easy-faststream queue status \
  --queue p_q.easy_faststream.test \
  --require-consumers \
  --watch 2
```
Use `--json` for machine-readable one-time checks.
Inspect a dead-letter queue
Inspection returns messages to the DLQ:
```bash
easy-faststream dlq inspect \
  --queue p_q.easy_faststream.test.dead \
  --limit 10
```
Do not run a normal subscriber on an operational replay DLQ. A subscriber may
ACK and remove messages before the CLI can inspect or replay them.
Dry-run a replay
```bash
easy-faststream dlq replay \
  --queue p_q.easy_faststream.test.dead \
  --exchange ex.easy_faststream.test \
  --routing-key easy.test \
  --limit 10 \
  --dry-run \
  --operator ek
```
Replay messages
```bash
easy-faststream dlq replay \
  --queue p_q.easy_faststream.test.dead \
  --exchange ex.easy_faststream.test \
  --routing-key easy.test \
  --limit 10 \
  --operator ek
```
The command asks for confirmation unless `--yes` is supplied.
Replay behavior:
Publish with RabbitMQ publisher confirms.
Require the destination to be routable.
ACK the DLQ message only after confirmed publishing.
Requeue the DLQ message when publishing fails.
Remove internal retry headers before republishing.
Preserve useful message and correlation metadata.
Replay audit records are written as JSON Lines to:
```text
logs/dlq_replay_audit.jsonl
```
Use `--audit-file` to choose another destination or `--no-audit` to disable
audit writing.
Development
Install the project:
```bash
python -m pip install -e ".[dev]"
```
Run linting and tests:
```bash
python -m ruff check src tests
python -m pytest
python -m compileall -q src
```
Build and validate distributions:
```bash
python -m build
python -m twine check dist/*
```
License
Licensed under the MIT License. See `LICENSE`.