Getting started

Install

pip install --extra-index-url https://acemq.org/pypi/simple/ acemq-amqp-fastapi

The extra index is needed because AceMQ is not on PyPI before 1.0. It brings acemq-amqp with the RabbitMQ transport, FastAPI and pydantic-settings with it — there is no second install to remember.

--extra-index-url rather than --index-url, because FastAPI, pydantic and aio_pika come from PyPI and only AceMQ comes from the AceMQ feed. Using --index-url replaces PyPI rather than adding to it, and the install then fails on the first ordinary dependency.

Python 3.10 or newer, which is the library's floor.

In a requirements.txt, the index goes in the file so nobody has to remember the flag:

--extra-index-url https://acemq.org/pypi/simple/
acemq-amqp-fastapi==0.1.5

Or, for a project with a pyproject.toml:

[project]
dependencies = ["acemq-amqp-fastapi>=0.1.5,<0.2"]

with the index configured where the install happens — PIP_EXTRA_INDEX_URL, a pip.conf, or [[tool.uv.index]]. A pyproject.toml cannot carry an index URL itself; that is pip's configuration, not the project's metadata.

This release wires acemq-amqp 0.7.12, the current library release, through a declared range of >=0.7.12,<0.8 — see the version table. Installing acemq-amqp-fastapi brings whichever release inside that range is newest, so there is nothing to pin yourself.

A broker

docker run -d --rm --name acemq -p 5672:5672 -p 15672:15672 rabbitmq:4-management

The whole application

# app.py
from typing import Annotated

from acemq_amqp import Ack, Message, Publisher, accept
from acemq_fastapi import AceMQ, AceMQSettings, publishes
from fastapi import Depends, FastAPI

acemq = AceMQ(
    AceMQSettings(
        url="amqp://guest:guest@localhost:5672/",
        topology={"queues": [{"name": "orders", "dead_letter": True}]},
    )
)


@acemq.consumer("orders")
async def handle_order(message: Message) -> Ack:
    print("handling", message.payload)
    return accept()


app = FastAPI(lifespan=acemq.lifespan)
app.include_router(acemq.health_router())

Orders = Annotated[Publisher, Depends(publishes("orders"))]


@app.post("/orders")
async def place(orders: Orders) -> dict[str, str]:
    result = await orders.send({"sku": "A-1"})
    return {"id": result.message_id}
uvicorn app:app
curl -XPOST localhost:8000/orders
curl localhost:8000/health/acemq

The POST publishes, the consumer picks it up on the same event loop, and the log shows the handler running before the response has finished being written.

The four lines that matter

acemq = AceMQ(...)                       # settings, and a registry. Connects to nothing.
@acemq.consumer("orders")                # a registration, held until startup
app = FastAPI(lifespan=acemq.lifespan)   # where the connection is opened and drained
Depends(publishes("orders"))             # a publisher, per request, from app.state

Nothing connects at import time. That is the point: app.py can be imported by a test, by alembic, or by --help without a broker existing.

Settings from the environment

The settings above can come from the environment instead, which is what a deployment will do:

export ACEMQ_URL=amqps://broker.internal:5671/
export ACEMQ_USERNAME=orders-service
export ACEMQ_PASSWORD=...
export ACEMQ_CONSUMER__PREFETCH=32
export ACEMQ_CONSUMER__SHUTDOWN_TIMEOUT=20
acemq = AceMQ()   # reads ACEMQ_*, and a .env file if one is beside the process

Every setting, and what each one does, is in configuration.

Then