FastAPI Lifespan Events Fix
In this tutorial, you'll learn about FastAPI Lifespan Events Fix. We cover key concepts, practical examples, and best practices.
The Problem
Database connections, Redis clients, and ML models need initialization before the first request and cleanup after the last. Without lifespan events, you manage these manually.
Quick Fix
Wrong — global init without lifecycle
from redis import asyncio as aioredis
# Redis client created at module import time
redis = aioredis.Redis(host='localhost')
@app.get("/items")
async def get_items():
await redis.get("key") # May fail if Redis wasn't ready
Output: If Redis is down during import, the server crashes. No clean shutdown — connections stay open.
Correct — startup and shutdown events
from contextlib import asynccontextmanager
from fastapi import FastAPI
from redis import asyncio as aioredis
redis: aioredis.Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
global redis
redis = aioredis.Redis(host='localhost')
await redis.ping()
print("Redis connected")
yield
# Shutdown
await redis.close()
print("Redis disconnected")
app = FastAPI(lifespan=lifespan)
Output: Redis connects on startup, disconnects on shutdown. Server doesn't start if Redis is unavailable.
Legacy decorator pattern
app = FastAPI()
@app.on_event("startup")
async def startup():
app.state.redis = aioredis.Redis(host='localhost')
print("Startup complete")
@app.on_event("shutdown")
async def shutdown():
await app.state.redis.close()
print("Shutdown complete")
Multiple lifespan tasks
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: initialize services
db = await create_db_pool()
redis = await create_redis()
ml_model = load_model()
app.state.db = db
app.state.redis = redis
app.state.ml_model = ml_model
yield
# Shutdown: close connections
await db.close()
await redis.close()
print("All resources cleaned up")
Accessing lifespan resources
from fastapi import Request
@app.get("/predict")
async def predict(request: Request, text: str):
model = request.app.state.ml_model
result = model.predict(text)
return {"prediction": result}
Error handling in lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
redis = aioredis.Redis(host='localhost')
await redis.ping()
app.state.redis = redis
yield
except ConnectionError:
print("Warning: Redis unavailable, running without cache")
app.state.redis = None
yield
finally:
if app.state.redis:
await app.state.redis.close()
Prevention
- Use the
lifespanparameter for startup/shutdown logic (modern approach). - Store shared resources in
app.state. - Handle initialization failures gracefully — allow the app to start with degraded functionality.
Common Mistakes with lifespan events
- Misunderstanding that
Stringis[Char]with poor performance for large text operations - Using
foldlinstead offoldl'causing stack overflow on large lists - Forgetting
deriving (Show, Eq)on custom data types needed for debugging
These mistakes appear frequently in real-world FASTAPI code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.
Practice Exercise
Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.
This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro