Skip to content

FastAPI and FastStream sharing a container

An HTTP API and a message consumer that run in one process can share one container, so both get the same APP-scoped instances, such as one connection pool. By default each integration opens the root container at startup and closes it at shutdown, so the first app to stop closes it under the other. Pass manage_lifespan=False to both setup_di calls and open and close the container yourself. The flag needs modern-di-fastapi 4.1.0 and modern-di-faststream 4.1.0 or later.

Setup

import dataclasses
import typing

import fastapi
import faststream
import modern_di_fastapi
import modern_di_faststream
from faststream.nats import NatsBroker
from modern_di import Container, Group, Scope, providers


@dataclasses.dataclass(kw_only=True, slots=True)
class ConnectionPool:
    closed: bool = False

    async def close(self) -> None:
        self.closed = True


class Dependencies(Group):
    pool = providers.Factory(
        ConnectionPool, scope=Scope.APP, cache=providers.CacheSettings(finalizer=ConnectionPool.close)
    )


container = Container(groups=[Dependencies])

app = fastapi.FastAPI()
modern_di_fastapi.setup_di(app, container, manage_lifespan=False)

broker = NatsBroker()
stream = faststream.FastStream(broker)
modern_di_faststream.setup_di(stream, container, manage_lifespan=False)
container.validate()


@app.get("/pool")
async def pool_status(
    pool: typing.Annotated[ConnectionPool, modern_di_fastapi.FromDI(ConnectionPool)],
) -> dict[str, bool]:
    return {"closed": pool.closed}


@broker.subscriber("orders")
async def handle_order(
    pool: typing.Annotated[ConnectionPool, modern_di_faststream.FromDI(ConnectionPool)],
) -> dict[str, bool]:
    return {"closed": pool.closed}

With the flag off, each integration still registers its context providers and builds a child container for every request or message. Only the root is left to you.

Running the apps side by side

Run the uvicorn server and the FastStream app in one asyncio.gather, inside async with container::

import asyncio

import uvicorn


async def main() -> None:
    server = uvicorn.Server(uvicorn.Config(app))
    async with container:
        await asyncio.gather(stream.run(), server.serve())


if __name__ == "__main__":
    asyncio.run(main())

Put stream.run() first. On SIGTERM uvicorn stops, puts back the signal handler it replaced at startup, and raises the signal again, so FastStream stops after it. In the other order, FastStream's handler replaces uvicorn's, and uvicorn keeps serving after FastStream has stopped.

The container closes when the async with block exits, after both apps have stopped. APP finalizers run once, at that point.

Starting FastStream inside the FastAPI lifespan

To run one uvicorn server, start the FastStream app from the FastAPI lifespan and enter the container there, around stream.start() and stream.stop():

import contextlib
from collections.abc import AsyncIterator

import fastapi
import modern_di_fastapi
import modern_di_faststream


@contextlib.asynccontextmanager
async def lifespan(_: fastapi.FastAPI) -> AsyncIterator[None]:
    async with container:
        await stream.start()
        try:
            yield
        finally:
            await stream.stop()


app = fastapi.FastAPI(lifespan=lifespan)
app.add_api_route("/pool", pool_status)
modern_di_fastapi.setup_di(app, container, manage_lifespan=False)
modern_di_faststream.setup_di(stream, container, manage_lifespan=False)

The container closes after stream.stop() returns, so messages still in flight at shutdown resolve from an open container. Since modern-di-fastapi 4.2.0, the FastAPI side at its default flag would give the same order, because it closes the container after your lifespan's shutdown code. Earlier releases closed it inside your lifespan, before stream.stop() ran.

stream.start() and stream.stop() do not run a FastStream(lifespan=...), so put that setup and teardown in the FastAPI lifespan.

Pitfalls

  • Set the flag on both integrations. One left at the default still closes the root when its app stops.
  • With the flag off, nothing else closes the root. Without your async with container: block, APP finalizers never run.

See also