Metadata-Version: 2.4
Name: metal-rabbit
Version: 0.4.0
Summary: RabbitMQ configuration library for FastAPI
License: MIT
Keywords: rabbitmq,aio-pika,asyncio,fastapi,python
Author: Naylson Ferreira
Author-email: naylsonfsa@gmail.com
Requires-Python: >=3.14
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.14
Provides-Extra: fastapi
Provides-Extra: otel
Requires-Dist: aio-pika (>=9.6.0,<10.0.0)
Requires-Dist: fastapi (>=0.115.12,<0.116.0) ; extra == "fastapi"
Requires-Dist: hard-lint-py (>=0.4.0,<0.5.0)
Requires-Dist: loguru (>=0.7.0,<0.8.0)
Requires-Dist: opentelemetry-api (>=1.20.0) ; extra == "otel"
Requires-Dist: opentelemetry-sdk (>=1.20.0) ; extra == "otel"
Requires-Dist: pydantic (>=2.0.0,<3.0.0)
Requires-Dist: pydantic-settings (>=2.2.1,<3.0.0)
Project-URL: Homepage, https://github.com/naylsonferreira/metal-rabbit
Project-URL: Issues, https://github.com/naylsonferreira/metal-rabbit/issues
Project-URL: Repository, https://github.com/naylsonferreira/metal-rabbit
Description-Content-Type: text/markdown

# 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

```bash
pip install metal-rabbit
```

For FastAPI integration (dependency and correlation-id middleware):

```bash
pip install metal-rabbit[fastapi]
```

For OpenTelemetry metrics on consumers:

```bash
pip install metal-rabbit[otel]
```

Neither extra is needed for publishing and consuming — see [Instrumentation](#instrumentation).

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:

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

## Quick start with FastAPI

```python
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.

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

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

You can also pass a URL string directly:

```python
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.

```python
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.

```python
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):

```python
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:

```python
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):

```python
@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:

```python
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.

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

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

| Instrumentation | What it does | Needs |
|---|---|---|
| `NullInstrumentation` | nothing; the default | — |
| `CorrelationInstrumentation` | injects the correlation id into published payloads, binds it from consumed ones | — |
| `MetricsInstrumentation` | counts handled messages by outcome and records the handler's duration | `metal-rabbit[otel]` |
| `CompositeInstrumentation` / `compose(...)` | runs several in order | — |

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

```python
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.

```python
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`):

| Instrument | Type |
|---|---|
| `messaging.consumed.messages` | counter |
| `messaging.consumed.duration` | histogram, seconds |

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:

```python
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

```python
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)`):

```python
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

```bash
# 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

