The lifespan

acemq.lifespan is where everything happens. It opens the connection, applies the topology, subscribes the registered consumers, hands control back to the application, and on the way out drains the consumers and releases the connection.

app = FastAPI(lifespan=acemq.lifespan)

What it does, in order

Startup is all-or-nothing. A consumer that cannot subscribe — a queue the login has no permission on, a broker that refuses a declaration — unwinds what has already been done and lets the exception out, so the process refuses to start rather than serving routes with no consumers behind them. A half-started application is worse than one that did not start: from outside they look the same.

Keeping the lifespan you already have

FastAPI takes one lifespan. compose_lifespans nests several:

from acemq_fastapi import compose_lifespans

@asynccontextmanager
async def database(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
    pool = await open_pool()
    try:
        yield {"pool": pool}
    finally:
        await pool.close()

app = FastAPI(lifespan=compose_lifespans(database, acemq.lifespan))

Left to right is outside in. database opens first and closes last; acemq.lifespan opens last and closes first.

Put AceMQ last. Consumers should start after the things their handlers use and stop before those go away. A handler reaching for a connection pool that has already been closed is the bug this ordering exists to prevent, and it only shows up during a deployment, in a log nobody is reading.

Whatever each lifespan yields as a mapping is merged and reaches the request state, in order, so a later one can override an earlier one's key. acemq.lifespan yields {"acemq": acemq}.

Write the cleanup in a try/finally rather than on a line after the yield. When a later lifespan fails on the way up, the exception is thrown into the earlier ones at their yield — a bare line after it never runs. acemq.lifespan is written this way for exactly that reason, and there is a test that would fail if it stopped being.

Except for what needs the connection

Put AceMQ last is about the things handlers use. Something that uses the connection — an OutboxRelay, a Requester, a Scheduler, a stream reader — has the opposite requirement, because acemq.connection raises until the lifespan has opened it. Those go to the right:

@asynccontextmanager
async def relay(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
    running = OutboxRelay(acemq.connection, store)
    running.start()
    try:
        yield {"relay": running}
    finally:
        await running.close()

app = FastAPI(lifespan=compose_lifespans(database, acemq.lifespan, relay))

Outside in: database, then AceMQ, then relay. The relay is built after the connection is open and closed before the connection goes away, and the handlers still have the pool either side of themselves.

The two rules collide when a handler needs one of those objects — a handler that asks another service a question needs a Requester, and nothing can be on both sides of the same yield. Patterns has the way through, which is acemq.consumer.auto_start=false and start_consumer.

Without an ASGI application at all

A worker process that consumes and serves no HTTP wants the same start-up, and should not have to construct a FastAPI application to get it:

async def main() -> None:
    await acemq.start()
    try:
        await stop_signal.wait()
    finally:
        await acemq.drain()
        await acemq.aclose()

start() takes an optional app; without one, nothing is put on app.state and the request dependencies are simply not available — which is correct, because there are no requests.

The connection is yours if you want it

AceMQ(connection_factory=...) replaces the one thing this package does that acemq-amqp already does better:

async def open_it(settings: AceMQSettings) -> Connection:
    return await acemq_amqp.connect(settings.broker_url, codec=my_codec, heartbeat=15)

acemq = AceMQ(connection_factory=open_it)

Everything else keeps working against it, because the registry, the dependencies and the health check take the connection rather than creating one. It is the same shape as the Spring starter's @ConditionalOnMissingBean AceMq: define your own and the rest of the wiring still applies.

Two applications in one process

Each gets its own. The dependencies find the integration through request.app.state.acemq, not through a module-level global, so a main application and a mounted admin one — or two applications in one test session — reach their own connection rather than whichever was built last.