LLM Best Practices

FastAPI: Lifespan Context Manager

4 min read

View markdown source

Overview

FastAPI’s lifespan context manager is the single point where application-scoped resources are created and destroyed. It replaces the deprecated @app.on_event("startup") and @app.on_event("shutdown") decorators. Resources opened before yield are available for the full life of the process; resources closed after yield are released on graceful shutdown. This page covers how to initialise a database pool, HTTP client, and ML model in lifespan, and how to expose them via app.state and dependency injection. See fastapi-dependencies for per-request resources.

Replace @app.on_event with an asynccontextmanager lifespan

The event-based decorators are deprecated in FastAPI 0.93+ and will be removed. The lifespan pattern is the replacement.

from contextlib import asynccontextmanager
from fastapi import FastAPI
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup: runs before the first request
    app.state.db = create_async_engine(settings.database_url, pool_size=10)
    yield
    # shutdown: runs after the last request, before the process exits
    await app.state.db.dispose()
 
app = FastAPI(lifespan=lifespan)

Everything before yield is startup. Everything after is shutdown. The context manager guarantees the shutdown block runs even if a startup step raises after partial initialisation.

Initialise the database connection pool in lifespan

Opening a connection pool in lifespan means one pool per worker, not one pool per request. The pool is shared across all requests handled by that worker process.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(
        settings.database_url,
        pool_size=10,
        max_overflow=5,
        pool_pre_ping=True,
    )
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()

pool_pre_ping=True tests each connection before use, discarding stale ones after a database restart. Set pool_size to match the expected concurrent requests per worker. See postgres for the connection limit math and fastapi-async-io for why async drivers are required.

Expose lifespan state through dependencies, not globals

Accessing app.state from route handlers directly works but couples routes to the application object. Wrap it in a dependency instead.

from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
 
def get_session_factory(request: Request):
    return request.app.state.session_factory
 
async def get_db(factory=Depends(get_session_factory)) -> AsyncGenerator[AsyncSession, None]:
    async with factory() as session:
        yield session

This keeps route functions unaware of where the factory lives and lets tests substitute it via app.dependency_overrides. See fastapi-dependencies for the full dependency chain.

Open shared HTTP clients in lifespan

An httpx.AsyncClient with connection pooling is more efficient than creating a new client per request. Initialise it once in lifespan and share it.

import httpx
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
        app.state.http_client = client
        yield
    # AsyncClient closes automatically when the async with block exits

Using async with for the client ensures the connection pool drains and sockets close on shutdown even if lifespan is interrupted. See fastapi-async-io for why httpx.AsyncClient is preferred over requests.

Load heavy resources once, before the first request

ML models, in-memory indexes, and compiled regex patterns are expensive to load. Load them in lifespan so the first request does not pay the cost.

import torch
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    model_path = settings.model_path
    app.state.classifier = torch.load(model_path, map_location="cpu")
    app.state.classifier.eval()
    yield
    del app.state.classifier

Loading in lifespan also makes failures visible at startup rather than on the first live request. A worker that cannot load a model should fail fast and let the process manager restart it rather than serving degraded traffic.

Handle partial startup failures cleanly

If lifespan raises before yield, FastAPI will not start the server. Structure startup steps so that failure is explicit and the partially-initialised state is released.

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(settings.database_url)
    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))  # fail fast if DB is unreachable
        app.state.engine = engine
        yield
    finally:
        await engine.dispose()

The finally block in the generator runs whether or not yield was reached, so the engine is always disposed. This avoids leaked connections when the startup health check fails. See python-async for the general pattern of cleanup in async context managers.

Graph View

Backlinks