RabbitMQ Python SOLID Template

This repository is a reference example of how to work safely with RabbitMQ from a Python service using FastStream, while keeping business logic, transport, cache, and infrastructure concerns separated.

The project demonstrates a worker-service approach where:

• FastStream is responsible only for RabbitMQ integration.
• RabbitMQ is used with durable queues, persistent messages, and DLX/TTL retries.
• Redis stores idempotency state for message processing.
• Application layer depends on interfaces, not on FastStream, Redis, or RabbitMQ.
• Domain layer contains business rules without infrastructure dependencies.
• Code follows SOLID principles: dependencies point inward, and transport details are replaceable.

Idempotency Algorithm

RabbitMQ provides at-least-once delivery semantics, so the same message can be delivered more than once. A safe consumer must treat duplicate delivery as a normal case, not as an exceptional situation.

This project implements idempotency with two Redis keys per event:

• idempotency:processing:{event_id} marks that a consumer is currently working on the event;
• idempotency:done:{event_id} marks that the event has already been processed successfully.

The processing algorithm is:

1. Check whether the done key exists. If it exists, the event was already processed and the consumer skips it.
2. Create a processing key with a unique token and a short TTL. This key acts as a distributed lock for the current event.
3. Check the done key again after acquiring the lock. This closes the race where another consumer may have completed the event between the first check and lock acquisition.
4. Run the business handler.
5. If the handler finishes successfully, write the done key with a longer TTL.
6. Remove the processing key only if its value is still equal to the current consumer token.
7. If the handler fails, do not write the done key. The exception is propagated, FastStream rejects the message, and RabbitMQ sends it through the retry flow.

The important detail is that a failed attempt does not mark the event as done. Only successful business execution does.

IdempotencyGuard

IdempotencyGuard is the application-level context manager that owns this algorithm:

async with IdempotencyGuard(cache=cache, event_id=event.event_id) as should_process:
    if not should_process:
        return

    await process_business_logic(event.payload)

Its responsibilities are intentionally narrow:

• check whether the event is already completed;
• acquire a processing lock before the business handler runs;
• expose should_process to the caller;
• write the done marker only after successful handler execution;
• safely release the processing lock after the block exits.

The guard depends on KeyValueCache[str], not on Redis directly:

class IdempotencyGuard(AbstractAsyncContextManager[bool]):
    def __init__(
        self,
        cache: KeyValueCache[str],
        event_id: UUID,
        *,
        processing_ttl: int = 300,
        done_ttl: int = 3600 * 24,
    ):
        ...

This keeps the idempotency logic inside the application layer and makes Redis an implementation detail.

Integrating A Consumer

To integrate idempotency into a new consumer:

1. Wrap the handler body with IdempotencyGuard.
2. Pass the event id from the event envelope.
3. Return immediately when should_process is False.
4. Keep business logic inside the guarded block.
5. Let exceptions propagate so RabbitMQ can retry the message.

Example:

class UserCreatedConsumer:
    def __init__(self, cache: KeyValueCache[str]):
        self._cache = cache

    async def process(self, event: Event[UserCreated]) -> None:
        async with IdempotencyGuard(self._cache, event.event_id) as should_process:
            if not should_process:
                return

            await self._create_user_projection(event.payload)

The FastStream subscriber should stay thin and delegate to the application consumer:

@rabbit_broker.subscriber(main_queue, ack_policy=AckPolicy.REJECT_ON_ERROR)
async def handle(event: Event[UserCreated]) -> None:
    await consumer.process(event)

This structure keeps idempotency independent from FastStream and reusable across different RabbitMQ consumers.

Quick Start

# Install dependencies
uv sync

# Prepare environment files
cp .env.example .env
cp rabbit.env.example rabbit.env

# Start RabbitMQ, Redis, and the worker
docker compose up -d --build --force-recreate

RabbitMQ Management UI will be available at http://localhost:15673.

Publish a test event from the worker container:

docker exec -it rps-worker python -m rabbitmq_python_solid.infrastructure.faststream.publish_important_message

Compile dependencies for the Docker image:

uv pip compile pyproject.toml -o src/requirements.txt

Why This Project Exists

Queue processing often looks simple: subscribe to a queue, receive a message, run a handler. In production, a worker must handle retries, crashes, duplicate deliveries, message durability, backpressure, and a clean separation between business logic and transport details.

This project demonstrates a baseline safe structure:

• messages are published as persistent;
• queues are declared as durable;
• the consumer uses prefetch_count;
• failed processing is not acknowledged as successful;
• retries are implemented through a dedicated queue with TTL and dead-letter routing;
• duplicate deliveries of the same event do not trigger duplicate business processing;
• the application handler does not know about FastStream decorators or RabbitMQ APIs.

Domain Layer

The domain layer contains business entities and rules that do not depend on the message broker.

class ImportantMessage(BaseModel):
    text: str
    author: str
    created_at: datetime

The importance check is isolated from the delivery mechanism:

class ImportanceChecker:
    @staticmethod
    def is_important(msg: ImportantMessage) -> bool:
        return bool(randint(0, 1))

In a real project, this layer can contain routing rules, validation, calculations, processing policies, and domain services. The important part is that it does not import FastStream, RabbitMQ, Redis, or environment settings.

Application Layer

The application layer contains use cases and interfaces through which business code communicates with the outside world.

Event Envelope

Messages are passed as an envelope with event_id and creation time, not as a bare payload:

class Event[T](BaseModel):
    event_id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    payload: T

event_id is used for idempotency. This is essential for RabbitMQ because a consumer must be ready for redelivery.

Publisher Interface

The application layer knows only the publishing abstraction:

class Publisher[T](ABC):
    @abstractmethod
    async def publish(self, message: Event[T]) -> None: ...

The FastStream implementation lives in the infrastructure layer. This allows use cases to be tested without RabbitMQ and lets the transport be replaced without rewriting business logic.

KeyValueCache Interface

Idempotency relies on an abstract key-value storage:

class KeyValueCache[T](ABC):
    @abstractmethod
    async def add(self, key: str, value: T, ttl: int) -> None: ...

    @abstractmethod
    async def delete_if_value(self, key: str, value: T) -> bool: ...

The concrete Redis implementation is hidden behind the interface. This preserves the Dependency Inversion Principle: the application layer does not depend on aiocache or Redis APIs.

Idempotent Consumer

The application consumer receives Event[ImportantMessage], not a transport message:

class ImportantMessageConsumer:
    def __init__(self, cache: KeyValueCache[str]):
        self._cache = cache

    async def process(self, msg: Event[ImportantMessage]) -> None:
        async with IdempotencyGuard(cache=self._cache, event_id=msg.event_id) as should_process:
            if not should_process:
                return

            is_important = ImportanceChecker.is_important(msg.payload)

IdempotencyGuard protects processing from duplicates:

• if the event is already completed, processing is skipped;
• before processing, a processing lock with TTL is created;
• after successful execution, a done marker is created;
• the lock is removed only by its owner through a compare-and-delete Lua script;
• on exception, the done marker is not created, so the message can be retried.

Infrastructure Layer

The infrastructure layer contains concrete technology details: FastStream, RabbitMQ, Redis, environment settings, and worker startup.

FastStream Broker

rabbit_broker = RabbitBroker(settings.rabbitmq.rabbitmq_uri)

The broker is created at the infrastructure level and does not leak into domain or application code.

RabbitMQ Queues

The main queue is durable and dead-letters failed messages into the retry queue:

main_queue = RabbitQueue(
    MAIN_QUEUE_NAME,
    durable=True,
    arguments={
        "x-dead-letter-exchange": "",
        "x-dead-letter-routing-key": RETRY_QUEUE_NAME,
    },
)

The retry queue holds the message for a defined time and then routes it back:

retry_queue = RabbitQueue(
    RETRY_QUEUE_NAME,
    durable=True,
    arguments={
        "x-message-ttl": 60_000,
        "x-dead-letter-exchange": "",
        "x-dead-letter-routing-key": MAIN_QUEUE_NAME,
    },
)

This approach avoids manual sleep calls inside the consumer and does not block processing of other messages.

Consumer Binding

The FastStream decorator stays at the infrastructure boundary:

@rabbit_broker.subscriber(
    main_queue,
    channel=Channel(prefetch_count=10),
    ack_policy=AckPolicy.REJECT_ON_ERROR,
    persistent=True,
    no_reply=True,
)
async def handle(event: Event[ImportantMessage]) -> None:
    await consumer.process(event)

Important settings:

• prefetch_count=10 limits the number of unacknowledged messages per consumer;
• AckPolicy.REJECT_ON_ERROR sends failed messages into the retry flow;
• persistent=True matches the durable approach to messages and queues;
• no_reply=True makes it explicit that this is event processing, not RPC.

Publisher

Publishing uses mandatory=True and persist=True:

await rabbit_broker.publish(message, queue=self._queue_name, mandatory=True, persist=True)

persist=True asks RabbitMQ to persist the message to disk for durable queues. mandatory=True helps avoid silently ignoring cases where a message cannot be routed to a queue.

Redis Idempotency

Redis is used as an infrastructure mechanism for duplicate-processing protection, not as business storage. The delete_if_value method removes the lock only if the stored value matches the current handler token:

if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
end

return 0

This protects against a stale consumer deleting a lock that already belongs to another process after TTL expiration.

SOLID In This Project

• Single Responsibility Principle: domain owns rules, application owns use cases, infrastructure owns RabbitMQ/Redis/FastStream integration.
• Open/Closed Principle: new consumers, publishers, or cache implementations can be added without rewriting existing business logic.
• Liskov Substitution Principle: Publisher and KeyValueCache implementations can be replaced with test or production adapters.
• Interface Segregation Principle: application code uses narrow interfaces, not large RabbitMQ or Redis clients.
• Dependency Inversion Principle: high-level scenarios depend on abstractions, while FastStream and Redis are implementation details.

Project Structure

src/rabbitmq_python_solid/
+-- domain/
|   +-- important_message.py           # Domain message model
|   +-- check_importance.py            # Domain importance rule
+-- application/
|   +-- consumers/
|   |   +-- important_message.py       # Message-processing use case
|   |   +-- idempotency_guard.py       # Duplicate-processing protection
|   +-- exceptions/                    # Application-layer errors
|   +-- interfaces/
|       +-- common/
|           +-- event.py               # Event envelope
|           +-- publisher.py           # Publishing interface
|           +-- key_value_cache.py     # Key-value storage interface
+-- infrastructure/
|   +-- aiocache/
|   |   +-- key_value_cache.py         # Redis adapter through aiocache
|   +-- faststream/
|       +-- app.py                     # FastStream app
|       +-- broker.py                  # RabbitBroker
|       +-- consumers.py               # FastStream subscriber bindings
|       +-- publisher.py               # FastStream publisher adapter
|       +-- rabbit_queues.py           # Durable queues and retry topology
|       +-- publish_important_message.py
|       +-- run_worker.py              # Worker entrypoint
+-- utils/
    +-- config/
    |   +-- settings.py                # Environment settings
    +-- logging/                       # Logging setup

Environment

Main environment variables are listed in .env.example:

RABBITMQ_HOST=rabbitmq
RABBITMQ_PORT=5672
RABBITMQ_USER=user
RABBITMQ_PASSWORD=password
RABBITMQ_IMPORTANT_MESSAGES_QUEUE=important-messages
RABBITMQ_IMPORTANT_MESSAGES_RETRY_QUEUE=important_messages.retry

REDIS_HOST=redis
REDIS_PORT=6379
REDIS_INDEX=0

RabbitMQ container variables are listed in rabbit.env.example.

Development

Formatting and linting:

uv run ruff format .
uv run ruff check . --fix

Run the worker stack:

docker compose up -d --build --force-recreate

Send a test event:

docker exec -it rps-worker python -m rabbitmq_python_solid.infrastructure.faststream.publish_important_message