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 (dependency and correlation-id middleware):

pip install metal-rabbit[fastapi]

For OpenTelemetry metrics on consumers:

pip install metal-rabbit[otel]

Neither extra is needed for publishing and consuming — see Instrumentation.

Requires Python 3.14+.

Configuration

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

.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).

Instrumentation

Every published message carries an AMQP timestamp. That is not for your code — it is what lets RabbitMQ answer queue_head_message_timestamp, the age of the oldest message still waiting. Without it that metric is NaN, and a dashboard cannot tell a queue that is merely busy from one that stopped being drained.

Anything beyond the timestamp is opt-in. The client takes an instrumentation object, applied in the two places worth wrapping: outgoing payloads (annotate) and the execution of a consumer handler (consuming). Passing nothing keeps the library exactly as quiet as before.

from metal_rabbit.client import MetalRabbitClient
from metal_rabbit.instrumentation import (
    CorrelationInstrumentation,
    MetricsInstrumentation,  # needs the [otel] extra
    compose,
)

client = MetalRabbitClient(
    instrumentation=compose(
        CorrelationInstrumentation(),
        MetricsInstrumentation(),
    )
)

Write your own by matching the Instrumentation protocol — two methods, no base class to inherit:

class TenantInstrumentation:
    def annotate(self, payload, *, queue):
        return payload

    @asynccontextmanager
    async def consuming(self, payload, *, queue, task):
        yield

Correlation id

One id follows a request across services: an HTTP request mints it, publishing carries it in the payload, consuming binds it back into the context, and the id ends up on every log line along the way.

from metal_rabbit.correlation import get_correlation_id, set_correlation_id
from metal_rabbit.fastapi import CorrelationIdMiddleware   # needs the [fastapi] extra

app.add_middleware(CorrelationIdMiddleware)   # honours X-Request-ID, echoes it back
logger.info("processing", correlation_id=get_correlation_id())

The payload field is request_id and the header is X-Request-ID; both are exported as CORRELATION_FIELD and CORRELATION_HEADER. With no id in context, get_correlation_id() returns "-" rather than None, so log lines keep their shape.

An incoming id is only adopted if it matches [A-Za-z0-9._-]{1,64}; anything else is replaced by a fresh one. The id arrives from outside — an HTTP header, or the payload of a message someone else published — and it ends up on every log line, so it is not a string to take on trust.

annotate does not touch the dict you hand it: it returns a copy carrying the id. A payload that already has a request_id keeps it, which is what lets an id survive a hop.

Metrics

MetricsInstrumentation records two instruments, both labelled with messaging.destination.name, task and outcome (ok or error):

The outcome starts as error and only becomes ok after the handler returns, so an exception is counted as a failure and still propagates.

Wire the duration view. OTel's default histogram buckets start at 0, 5, 10, 25 — sized for milliseconds, while the instrument reports seconds as the convention asks. A handler taking single-digit milliseconds lands entirely in the first bucket and histogram_quantile interpolates inside it: measured, a p95 of 2.6 ms worth of work read 4.75 s. The library ships the boundaries it expects:

from metal_rabbit.instrumentation import duration_view

MeterProvider(resource=resource, views=[duration_view()], metric_readers=[reader])

Pass meter= to the instrumentation if you keep your own provider instead of the global one.

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