Overview

A tutorial project implementing basic RabbitMQ use cases in Python:

• Tasks queue:
  • a single queue with one publisher and several consumers;
  • consumers implement retries on simulated failures and graceful shutdown;
• Routing: a message is published to several queues using CC headers and processing by different consumers;
• Dead letter exchange (dlx):
  • published messages are provided with a flag which toggles successful / failed consumer processing;
  • failed messages are redirected to a dead letter exchange and processed by another consumer;
• Idempotency and robustness (idm_rob):
  • at least once message publishing via transactional outbox & periodic publishing task;
  • publishing task schedule uses backoffs & jitter;
  • idempotent message consumption with simulated retries;
  • robust connections on publisher & consumer sides.

Implementation Details

• A FastAPI app is used as a publisher;
• each example is mostly independent from others, relying on its own RabbitMQ queues & exchanges, publisher and consumer connections, consumer logic, etc. (some general publisher & consumer logic is shared, though);
• messages and processing logs are stored in a PostgreSQL database;
• published messages are processed by one or more consumers, running as standalone services;
• project contains e2e tests for each example and a web UI for manual testing.

Stack

• Python 3.14;
• RabbitMQ 4 / aio-pika;
• FastAPI;
• PostgreSQL 17 / SQLAlchemy 2 / Alembic;
• Pydantic + Pydantic Settings;
• APScheduler (periodic tasks);
• pytest;
• Docker + Docker Compose.

Project structure

├── src/
│   ├── main.py            # Publiher entrypoint
│   ├── app.py             # FastAPI application factory
│   ├── config.py          # Loads and validates .env
│   ├── db/                # ORM models, migrations, repository & DB admin scripts
│   ├── middleware/
│   ├── models/            # Pydantic models
│   ├── rabbitmq/          # broker setup (queues/, configure.py) and consumer logic
│   ├── routes/            # /api (interact with db tables), /examples (trigger publishing) and /ui route handlers
│   ├── tasks/             # APScheduler periodic tasks
│   ├── ui/                # web UI implementation
│   └── util.py            # shared utility functions
|
├── tests/
│   ├── conftest.py        # test fixtures: fresh DB + virtual host per module, test FastAPI app & clients, consumer runner
│   ├── utils/             # helpers: consumer runner, data generators, db operations
│   └── tests/<example>/   # end-to-end tests per example
|
├── docs/                  # task plan and per-example specs
├── docker-compose.yml     # Postgres, RabbitMQ, API and consumers
├── Dockerfile.dev
├── pyproject.toml         # dependencies and pytest settings
└── .env.example           # configuration template (copy to .env)

Development setup

# Create project config
cp .env.example .env

# Run DB migrations & broker configuration in one-shot contatiners,
# then run the whole setup (PostgresSQL, RabbitMQ, publisher app and 3 consumers).
docker compose up -d

Testing

# Start the required infrastructure
docker compose up -d db rabbitmq

# Install dependencies
uv sync

# Run the tests from the host
uv run pytest

Web UI

Available at http://localhost:8080 (or whatever BACKEND__PORT is set to).