Metadata-Version: 2.4
Name: metal-rabbit
Version: 0.3.1
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
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: 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:

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

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

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

