Lifespan and Startup/Shutdown
Every production FastAPI app needs initialization at startup (connect to DB, warm up pool, load config) and cleanup at shutdown (close connections, flush buffers). The modern way is the lifespan context manager, which replaced the deprecated @app.on_event handlers.
Learning Focus
By the end of this lesson you can: implement a lifespan context manager for full startup/shutdown control, store shared resources on app.state, migrate from legacy on_event handlers, and test lifespan logic.
Modern Lifespan Pattern
app/main.py
from contextlib import asynccontextmanager
import httpx
import redis.asyncio as aioredis
from fastapi import FastAPI
from app.db.session import engine, AsyncSessionFactory
from app.core.config import settings
from sqlalchemy import text
@asynccontextmanager
async def lifespan(app: FastAPI):
# ===== STARTUP =====
# 1) Verify database connection
async with AsyncSessionFactory() as db:
await db.execute(text("SELECT 1"))
# 2) Initialize shared HTTP client
app.state.http_client = httpx.AsyncClient(timeout=10.0)
# 3) Initialize Redis client
app.state.redis = aioredis.from_url(settings.REDIS_URL, decode_responses=True)
# 4) Load any cached config from Redis
# await warm_up_cache(app.state.redis)
yield # App is now running and serving requests
# ===== SHUTDOWN =====
await app.state.http_client.aclose()
await app.state.redis.aclose()
await engine.dispose()
app = FastAPI(lifespan=lifespan)
Accessing app.state in Routes
app/routers/external.py
from fastapi import APIRouter, Request
import httpx
router = APIRouter()
@router.get("/github/user/{username}")
async def get_github_user(username: str, request: Request) -> dict:
client: httpx.AsyncClient = request.app.state.http_client
resp = await client.get(f"https://api.github.com/users/{username}")
resp.raise_for_status()
return resp.json()
Migrating from Legacy on_event
Legacy pattern (deprecated — still works but no longer recommended):
legacy-pattern.py
# ❌ Deprecated — use lifespan instead
@app.on_event("startup")
async def startup():
app.state.db = create_engine(...)
@app.on_event("shutdown")
async def shutdown():
await app.state.db.dispose()
Modern equivalent:
app/main.py
# ✅ Current pattern
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.db = create_engine(...)
yield
await app.state.db.dispose()
app = FastAPI(lifespan=lifespan)
Multiple Startup Tasks
app/main.py
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Run independent startup tasks concurrently
await asyncio.gather(
verify_db_connection(),
warm_redis_cache(),
load_feature_flags(),
)
yield
# Sequential cleanup
await app.state.http_client.aclose()
await engine.dispose()
Lifespan for Multiple Routers
Use a separate lifespan per router for modular initialization:
app/routers/payments.py
from contextlib import asynccontextmanager
import stripe
from fastapi import APIRouter
@asynccontextmanager
async def payment_lifespan(app: FastAPI):
from app.core.config import settings
stripe.api_key = settings.STRIPE_SECRET_KEY
yield
router = APIRouter(lifespan=payment_lifespan)
Common Pitfalls
| Pitfall | Cause / Symptom | Fix |
|---|---|---|
on_event deprecation warning | Using @app.on_event | Migrate to lifespan context manager |
app.state not available in routes | Accessing state before lifespan runs | Only access app.state after yield in lifespan |
| Startup failure crashes silently | Exception in lifespan before yield | Let it propagate — Uvicorn will log and exit |
| Resource leak on exception | Startup exception skips shutdown | Python asynccontextmanager handles this; teardown still runs |
| Startup blocks request handling | Long blocking operation in lifespan | Move slow startup to asyncio.gather for concurrency |
Hands-On Practice
app/main.py
from contextlib import asynccontextmanager
import time
import httpx
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
print(f"[startup] App starting at {time.strftime('%Y-%m-%d %H:%M:%S')}")
app.state.started_at = time.time()
app.state.http = httpx.AsyncClient()
yield
print(f"[shutdown] App stopping")
await app.state.http.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/uptime")
async def uptime(request: Request) -> dict:
elapsed = time.time() - request.app.state.started_at
return {"uptime_seconds": round(elapsed, 2)}
test-lifespan.sh
uvicorn app.main:app --reload
curl http://localhost:8000/uptime
# → {"uptime_seconds": 3.14}