It lets you run code once at startup (before requests are served) and once at shutdown (after the server stops accepting requests). It’s the modern replacement for the deprecated @app.on_event(“startup”/”shutdown”) decorators.
Typical uses: opening/closing a DB connection pool, loading an ML model, connecting to Redis, starting a background scheduler etc.
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# --- STARTUP ---
print("Starting up...")
# startup code
yield
# --- SHUTDOWN ---
print("Shutting down...")
# shutdown code
app = FastAPI(lifespan=lifespan)@asynccontextmanager
async def lifespan(app: FastAPI):
db = await connect_db()
cache = await connect_cache()
try:
yield
finally:
await db.close()
await cache.close()Use app.state to attach objects created at startup so routes can access them.
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import httpx
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http_client = httpx.AsyncClient()
yield
await app.state.http_client.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/ping")
async def ping(request: Request):
client = request.app.state.http_client
resp = await client.get("https://example.com")
return {"status": resp.status_code}| Rule | Detail |
|---|---|
| One mechanism only | If you pass lifespan, the old @app.on_event() handlers are ignored — pick one |
| Main app only | Lifespan runs only for the top-level app, not for mounted sub-applications |
| Runs once per process | With multiple Uvicorn workers, lifespan runs separately in each worker process |
| Not a request hook | It runs once for the whole app lifetime, not per-request (use middleware for per-request logic) |
| Must yield exactly once | The generator needs a single yield splitting startup from shutdown code |
Before lifespan, startup/shutdown logic used two separate decorators:
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def startup():
app.state.db = await connect_db()
@app.on_event("shutdown")
async def shutdown():
await app.state.db.close()This worked, but had real technical limitations in production apps: