Skip to content

RabbitMQ Integration

Add RabbitMQ helpers to an existing PolePosition project:

polepos add integration rabbitmq

The command creates:

src/<package>/integrations/rabbitmq/
  __init__.py
  consumer.py
  factory.py
  publisher.py
  schemas.py
  testing.py

It also updates:

  • src/<package>/settings.py
  • .env.example
  • pyproject.toml

Dependency

The command adds:

aio-pika>=9.0.0

Sync dependencies after adding the integration:

uv sync

Settings

Review the RabbitMQ values in .env:

RABBITMQ_ENABLED=false
RABBITMQ_URL=amqp://guest:guest@localhost:5672/
RABBITMQ_CLIENT_ID=<package>
RABBITMQ_EXCHANGE=<package>.events
RABBITMQ_EXCHANGE_TYPE=topic
RABBITMQ_EXCHANGE_DURABLE=true
RABBITMQ_DEFAULT_ROUTING_KEY=<package>.event
RABBITMQ_DEFAULT_QUEUE=<package>.events
RABBITMQ_QUEUE_DURABLE=true
RABBITMQ_PREFETCH_COUNT=10

Use the Publisher

The generated factory builds a publisher from settings. Keep connection and publisher lifetime management explicit in your runtime code.

Use the generated schemas to keep message payloads predictable.

Consumers and Workers

The generated consumer helper is scaffolding for a worker surface. It is not started automatically by the FastAPI app.

For production, run consumers as explicit worker processes or jobs so API startup remains fast and predictable.

Testing

Use InMemoryRabbitMQEventPublisher from testing.py when unit tests should assert published messages without connecting to RabbitMQ.

Validate

polepos check

The check command validates RabbitMQ files, settings, env values, and dependency signals without connecting to RabbitMQ.