Lifespan Events in FastAPI

Lifespan — manage application startup and shutdown (DB initialization, ML models, connection pools).

Modern Approach (FastAPI 0.95+)

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: runs when the application starts
    print("Starting application...")
    await init_db()
    yield
    # Shutdown: runs when the application stops
    print("Stopping application...")
    await close_db()

app = FastAPI(lifespan=lifespan)

Usage Examples

Database Initialization

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create tables on startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Close the connection pool
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Loading an ML Model

ml_model = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the model once on startup
    ml_model["classifier"] = load_model("model.pkl")
    yield
    # Clean up
    ml_model.clear()

app = FastAPI(lifespan=lifespan)

@app.post("/predict/")
def predict(data: PredictRequest):
    result = ml_model["classifier"].predict(data.features)
    return {"result": result}

Redis and HTTP Client

import httpx
import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await redis.from_url("redis://localhost")
    app.state.http_client = httpx.AsyncClient()
    yield
    await app.state.redis.close()
    await app.state.http_client.aclose()

Accessing State in Endpoints

@app.get("/cache/{key}")
async def get_cache(key: str, request: Request):
    value = await request.app.state.redis.get(key)
    return {"value": value}

Deprecated Approach (startup/shutdown events)

# Old approach — works, but not recommended
@app.on_event("startup")
async def startup():
    await init_db()

@app.on_event("shutdown")
async def shutdown():
    await close_db()