easy-faststream 0.2.0

A schema-first FastStream framework with automatic retries and dead-letter handling.

easy-faststream

easy-faststream is a schema-first wrapper around FastStream for RabbitMQ. Application developers define a Pydantic schema and a consumer function; the framework handles validation, retry, acknowledgements, and dead-letter routing.

Example

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


app = StreamApp.from_env()


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

Run the application through FastStream:

faststream run app:app.app

Environment variables

EASY_STREAM_RABBITMQ_URL=amqp://guest:guest@localhost:5672/
EASY_STREAM_APP_NAME=trip-consumer
EASY_STREAM_DEFAULT_EXCHANGE=easy.events
EASY_STREAM_DLQ_EXCHANGE=easy.events.dead
EASY_STREAM_MAX_RETRIES=3
EASY_STREAM_RETRY_DELAY_SECONDS=1
EASY_STREAM_RETRY_BACKOFF=2

Failure behavior

• Invalid schema: publish the original payload and validation details to the DLQ.
• Processing error: retry according to the configured policy, then publish to DLQ.
• NonRetryableError: skip retries and publish directly to DLQ.
• DLQ publish error: NACK and requeue the original RabbitMQ message.
• Success or successful DLQ publish: ACK the original message.