Some resources must exist before the app serves requests and be released after the last one: connection pools, HTTP clients, ML models, caches, background workers. The lifespan context manager is FastAPI's modern way to run setup and teardown code once per process.
from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# ---- startup ----
app.state.http = httpx.AsyncClient() # shared HTTP client
ml_models["sentiment"] = load_model() # heavy one-time work
print("Application ready")
yield
# ---- shutdown ----
await app.state.http.aclose()
ml_models.clear()
print("Shutting down")
app = FastAPI(lifespan=lifespan)
@app.get("/predict/")
async def predict(text: str):
async with app.state.http as client: # reuse the pooled client
resp = await client.get("https://api.example.com/enrich")
return {"sentiment": ml_models["sentiment"].score(text)}
yield runs once at startup, before the server accepts requests; everything after runs at shutdown.yield), the app refuses to start — fail-fast instead of serving with broken state.app.state is the conventional place to stash objects created here; alternatives are module-level globals (as with ml_models).@app.on_event("startup")
async def startup():
...
@app.on_event("shutdown")
async def shutdown():
...
on_event still works but is deprecated in favor of lifespan (since FastAPI 0.93). New code should always use the context manager — it guarantees that teardown runs even if startup partially completed, and it reads as one coherent block.
httpx.AsyncClient, warm caches, load ML models, start metric exporters, run lightweight migrations.engine.dispose()), flush logs/metrics, stop consumers, release locks.ready only once startup finished.# BAD: a new TCP handshake + pool per request
@app.get("/proxy/")
async def proxy(url: str):
async with httpx.AsyncClient() as client: # created and destroyed every time
return (await client.get(url)).json()
# GOOD: one client created in lifespan, pooled connections
@app.get("/proxy/")
async def proxy(url: str):
r = await app.state.http.get(url)
return r.json()
Creating a client per request wastes connection pools and TLS handshakes. Creating it in lifespan and closing it at shutdown is the correct pattern (the same reasoning as SQLAlchemy's module-level engine).
A: Once per worker process. uvicorn --workers 4 runs lifespan 4 times — each worker builds its own pools/models. Keep startup work idempotent and per-process-safe (don't run destructive migrations there; use Alembic in the deploy pipeline instead).
A: The lifespan function is async; call blocking code carefully — wrap heavy blocking init with await asyncio.to_thread(...) so the event loop is not frozen during startup, or accept the delay (nothing else is being served yet).
A: TestClient(app) used as a context manager triggers the lifespan: with TestClient(app) as client: .... Plain TestClient(app) without the with block skips startup/shutdown — a common reason state like app.state.http is missing in tests.
The lifespan context manager is the cleanest way to own process-wide resources: async setup before yield, guaranteed teardown after. It replaces on_event, integrates with testing via the with TestClient form, and sets the stage for the deployment patterns of the advanced level.