FastAPI: Lifespan, state & shared resources

Some things should exist exactly once per process: the database engine and its connection pool, an HTTP client with keep-alive connections, a loaded ML model, a validated settings object. The lifespan is an async context manager that runs code before the first request and after the last one. Whatever it creates can be handed to every request as state. Getting this right removes a whole family of production bugs: connection storms, leaked sockets, and models reloaded on every request.

1. The lifespan context manager

create once, share with every request, close at shutdown

from contextlib import asynccontextmanager
from typing import Annotated

import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.testclient import TestClient

events = []

def fake_upstream(request: httpx.Request) -> httpx.Response:        # stands in for another service
    return httpx.Response(200, json={"rate": 83.21, "path": request.url.path})

@asynccontextmanager
async def lifespan(app: FastAPI):
    events.append("startup: opening HTTP client (one pool of keep-alive connections)")
    http = httpx.AsyncClient(transport=httpx.MockTransport(fake_upstream), base_url="https://fx.example",
                             timeout=httpx.Timeout(5.0, connect=2.0))         # ALWAYS set timeouts
    model = {"name": "fraud-v3", "loaded": True}                               # e.g. an ML model, loaded once
    yield {"http": http, "model": model}                 # this dict becomes request.state
    await http.aclose()
    events.append("shutdown: HTTP client closed")

app = FastAPI(lifespan=lifespan)

def get_http(request: Request) -> httpx.AsyncClient:
    return request.state.http

@app.get("/convert/{amount}")
async def convert(amount: float, request: Request, http: Annotated[httpx.AsyncClient, Depends(get_http)]):
    rate = (await http.get("/usd-inr")).json()["rate"]
    return {"inr": round(amount * rate, 2), "model": request.state.model["name"]}

with TestClient(app) as client:                          # 'with' runs startup and shutdown
    events.append(f"request: {client.get('/convert/10').json()}")
    events.append(f"request: {client.get('/convert/2.5').json()}")
print("\n".join(events))

Output

startup: opening HTTP client (one pool of keep-alive connections)
request: {'inr': 832.1, 'model': 'fraud-v3'}
request: {'inr': 208.02, 'model': 'fraud-v3'}
shutdown: HTTP client closed

2. Three lifespan traps, demonstrated

TestClient without 'with', on_event next to lifespan, and a failing startup

import warnings
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

log = []

@asynccontextmanager
async def lifespan(app):
    log.append("lifespan startup")
    yield {"db": "engine"}
    log.append("lifespan shutdown")

app = FastAPI(lifespan=lifespan)
with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")

    @app.on_event("startup")                                  # the old API
    async def legacy_startup():
        log.append("on_event startup")
print("1. on_event:", caught[0].category.__name__, "-", str(caught[0].message).split(".")[0].strip())

@app.get("/")
def root(request: Request):
    return {"db": getattr(request.state, "db", "MISSING")}

print("2. without 'with':", TestClient(app).get("/").json())       # lifespan never ran
with TestClient(app) as client:
    print("   with 'with':   ", client.get("/").json())
print("   log:", log, "<- the on_event handler never ran")

@asynccontextmanager
async def broken(app):
    raise ConnectionError("database db-primary:5432 unreachable")
    yield

try:
    with TestClient(FastAPI(lifespan=broken)):
        pass
except ConnectionError as e:
    print("3. startup failure stops the app before any request:", e)

Output

1. on_event: DeprecationWarning - on_event is deprecated, use lifespan event handlers instead
2. without 'with': {'db': 'MISSING'}
   with 'with':    {'db': 'engine'}
   log: ['lifespan startup', 'lifespan shutdown'] <- the on_event handler never ran
3. startup failure stops the app before any request: database db-primary:5432 unreachable

3. HTTP clients: the most common resource bug

4. Graceful shutdown

When a deployment replaces a process, the server receives SIGTERM. uvicorn stops accepting new connections, lets in-flight requests finish (bounded by --timeout-graceful-shutdown), then runs your lifespan’s shutdown code. Two consequences:

Recap