metal-rabbit 0.3.1

RabbitMQ configuration library for FastAPI

metal-rabbit

Async RabbitMQ client library for FastAPI. Uses a single connect_robust connection with an async channel pool per process, decorator-based consumer API, and typed topology configuration via Pydantic.

Installation

pip install metal-rabbit

For FastAPI integration:

pip install metal-rabbit[fastapi]

Requires Python 3.14+.

Configuration

Settings are loaded from environment variables (or a .env file via pydantic-settings):

Variable	Default	Description
RABBITMQ_HOST	—	Broker hostname (required)
RABBITMQ_PORT	5672	Broker port
RABBITMQ_USER	—	Username (required)
RABBITMQ_PASSWORD	—	Password (required)
RABBITMQ_VHOST	/	Virtual host
RABBITMQ_HEARTBEAT	600	Heartbeat interval in seconds
RABBITMQ_BLOCKED_CONNECTION_TIMEOUT	300	Blocked connection timeout in seconds
RABBITMQ_CHANNEL_POOL_SIZE	10	Max channels in the pool
RABBITMQ_POOL_ACQUIRE_TIMEOUT	5	Seconds to wait for a free channel

.env example:

RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

Quick start with FastAPI

from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI
from metal_rabbit.client import MetalRabbitClient
from metal_rabbit.fastapi import get_metal_rabbit_client
from metal_rabbit.producer import ExchangeType, MetalRabbitConfig
from metal_rabbit.settings import MetalRabbitSettings

settings = MetalRabbitSettings()
client = MetalRabbitClient(settings=settings)

ORDERS_CONFIG = MetalRabbitConfig(
    exchange_name="orders",
    queue_name="orders.created",
    routing_key="orders.created",
    exchange_type=ExchangeType.DIRECT,
)


@client.consumer(ORDERS_CONFIG)
async def handle_order(payload: dict) -> None:
    print(f"order received: {payload}")


@asynccontextmanager
async def lifespan(app: FastAPI):
    await client.start(app)   # connects, starts consumer tasks, registers on app.state
    yield
    await client.close()      # graceful shutdown


app = FastAPI(lifespan=lifespan)


@app.post("/orders", status_code=202)
async def create_order(
    body: dict,
    rabbit_client: MetalRabbitClient = Depends(get_metal_rabbit_client),
) -> dict:
    await rabbit_client.publish(queue_name=ORDERS_CONFIG.queue_name, message=body)
    return {"status": "queued"}

Core concepts

MetalRabbitClient

The central object. Manages a single connect_robust connection (auto-reconnect) and an async channel pool. All methods are coroutines.

from metal_rabbit.client import MetalRabbitClient
from metal_rabbit.settings import MetalRabbitSettings

client = MetalRabbitClient(settings=MetalRabbitSettings())

You can also pass a URL string directly:

client = MetalRabbitClient(url="amqp://guest:guest@localhost/")

MetalRabbitConfig

Pydantic model that describes a topology (exchange + queue + binding). Fields are validated at construction time — empty or blank strings are rejected.

from metal_rabbit.producer import ExchangeType, MetalRabbitConfig

config = MetalRabbitConfig(
    exchange_name="events",
    queue_name="events.user_signed_up",
    routing_key="events.user_signed_up",
    exchange_type=ExchangeType.TOPIC,   # DIRECT | FANOUT | TOPIC | HEADERS
    durable=True,
    persistent=True,
)

ExchangeType

Enum for all supported RabbitMQ exchange types. Always use this instead of raw strings.

from metal_rabbit.producer import ExchangeType

ExchangeType.DIRECT   # "direct"
ExchangeType.FANOUT   # "fanout"
ExchangeType.TOPIC    # "topic"
ExchangeType.HEADERS  # "headers"

Publishing messages

Simple publish — directly to a queue (default exchange, no topology setup required):

await client.publish(queue_name="jobs", message={"job_id": "abc"})

Accepts dict, list, str, or bytes. Dicts and lists are serialized as JSON with content-type: application/json.

Producer — publish through a named exchange with full topology:

from metal_rabbit.producer import MetalRabbitProducer

producer = MetalRabbitProducer(client=client, config=config)
await producer.setup()                          # declares exchange, queue, and binding
await producer.publish({"event": "user.created", "id": 1})

Consuming messages

Decorator API (recommended with FastAPI):

@client.consumer(config)
async def handle_event(payload: dict) -> None:
    ...

await client.start()  # spawns one asyncio Task per registered consumer

Sync handlers are also accepted — the decorator detects coroutines automatically.

Class-based — subclass MetalRabbitConsumer for more control:

from metal_rabbit.consumer import MetalRabbitConsumer
import aio_pika

class OrderConsumer(MetalRabbitConsumer):
    config = MetalRabbitConfig(
        exchange_name="orders",
        queue_name="orders.created",
        routing_key="orders.created",
    )

    async def _on_message(self, message: aio_pika.IncomingMessage, payload: dict) -> None:
        print(payload)

consumer = OrderConsumer(client)
await consumer.start()
# ...
await consumer.stop()

Messages that fail to deserialize as JSON are nack'd and discarded. Unhandled exceptions in the callback trigger a nack (no requeue).

Low-level topology management

await client.declare_exchange(exchange_name="events", exchange_type="topic")
await client.declare_queue("events.user_signed_up")
await client.bind_queue(
    queue_name="events.user_signed_up",
    exchange_name="events",
    routing_key="events.user_signed_up",
)

ensure_direct_topology(config) does all three in one call and caches the result so subsequent calls are no-ops.

FastAPI dependency

get_metal_rabbit_client is a FastAPI dependency that reads the client from app.state.rabbitmq_client (set automatically by await client.start(app)):

from metal_rabbit.fastapi import get_metal_rabbit_client

@app.get("/healthcheck")
async def healthcheck(client: MetalRabbitClient = Depends(get_metal_rabbit_client)):
    await client.ping()
    return {"status": "ok"}

Development

# Install with dev dependencies
make dev

# Run unit tests
make test

# Run e2e tests (requires RabbitMQ)
docker compose up -d
make test-e2e

# Lint
make lint

# Auto-fix lint/format issues
make fix

License

MIT