Question 1 / 15
How do you implement caching with Redis in FastAPI?
Redis caching in FastAPI involves storing serialized responses or computed values in Redis with a TTL (time-to-live). The pattern is: check cache first, return if hit, otherwise compute, store, and return.
Integration approach:
- Create a Redis client in lifespan and store it on
app.state - Build a cache dependency or decorator
- Use JSON serialization for structured data, or raw bytes for simple values
Common patterns:
- Read-through cache: Check cache, miss -> compute -> store -> return
- Cache-aside: Application manages cache explicitly
- Write-through: Update cache when data changes
- Cache invalidation: Delete cache keys when underlying data changes
Libraries:
redis.asyncio(official async Redis client, formerlyaioredis)fastapi-cache2(decorator-based caching with Redis/Memcached backends)
Key design decisions:
- Key naming: Use a consistent scheme like
resource:id:field - TTL: Balance freshness vs performance (60s-3600s common)
- Serialization:
json.dumps()for dicts,picklefor complex objects (security risk)
Pseudo Code
# Cache-aside pattern
async def get_user(user_id):
cached = await redis.get(f"user:{user_id}")
if cached:
return json.loads(cached)
user = await db.fetch_user(user_id)
await redis.set(f"user:{user_id}", json.dumps(user), ex=300)
return userCode
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request
from typing import Annotated
import redis.asyncio as redis
import json
import hashlib
import functools
import logging
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Create Redis connection pool at startup
pool = redis.ConnectionPool.from_url(
"redis://localhost:6379",
max_connections=20,
decode_responses=True,
)
app.state.redis = redis.Redis(connection_pool=pool)
yield
await app.state.redis.aclose()
await pool.aclose()
app = FastAPI(lifespan=lifespan)
# --- Dependency-based caching ---
async def get_redis(request: Request) -> redis.Redis:
return request.app.state.redis
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
cache: Annotated[redis.Redis, Depends(get_redis)],
):
cache_key = f"user:{user_id}"
# Try cache first
cached = await cache.get(cache_key)
if cached:
return {"data": json.loads(cached), "source": "cache"}
# Cache miss: fetch from database
user = {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}
# Store in cache with 5-minute TTL
await cache.set(cache_key, json.dumps(user), ex=300)
return {"data": user, "source": "database"}
@app.put("/users/{user_id}")
async def update_user(
user_id: int,
cache: Annotated[redis.Redis, Depends(get_redis)],
):
# Update database...
user = {"id": user_id, "name": "Updated"}
# Invalidate cache (write-through)
await cache.delete(f"user:{user_id}")
return {"data": user, "status": "updated"}
# --- Decorator-based caching ---
def cached(ttl_seconds: int = 300, prefix: str = ""):
"""Decorator that caches endpoint responses in Redis."""
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
# Build cache key from function name + args
request = kwargs.get("request") or args[0]
key_data = f"{prefix or func.__name__}:{request.url.path}:{request.url.query}"
cache_key = hashlib.md5(key_data.encode()).hexdigest()
redis_client = request.app.state.redis
cached_data = await redis_client.get(cache_key)
if cached_data:
return json.loads(cached_data)
result = await func(*args, **kwargs)
await redis_client.set(
cache_key, json.dumps(result), ex=ttl_seconds
)
return result
return wrapper
return decorator
@app.get("/products")
@cached(ttl_seconds=60, prefix="products")
async def list_products(request: Request, category: str = "all"):
# Expensive query
return {"products": [{"id": 1, "name": "Widget"}], "category": category}
# --- Bulk cache operations ---
@app.get("/users")
async def list_users(
cache: Annotated[redis.Redis, Depends(get_redis)],
):
# Pipeline for multiple cache reads
async with cache.pipeline(transaction=False) as pipe:
for user_id in range(1, 11):
pipe.get(f"user:{user_id}")
results = await pipe.execute()
users = [
json.loads(r) for r in results if r is not None
]
return {"users": users, "cached_count": len(users)}Key Points
- •Create Redis client in lifespan with a connection pool for efficiency
- •Cache-aside pattern: check cache, miss -> compute -> store with TTL
- •Invalidate cache on writes to prevent stale data
- •Use Redis pipelines for bulk cache operations to reduce round trips
Gotchas
- •Cache stampede: many requests miss cache simultaneously -- use locking or staggered TTLs
- •JSON serialization cannot handle datetime, set, or custom objects without a custom encoder
- •Redis connection failures should be handled gracefully -- fall back to database, do not crash the endpoint
- •decode_responses=True returns strings, not bytes -- be consistent across your app