FastAPI Cheatsheet 17 — Caching: HTTP, Redis, fastapi-cache

@app.get("/posts")
async def posts(response: Response):
    response.headers["Cache-Control"] = "public, max-age=60, s-maxage=120"
    return await db.list_posts()

import hashlib

@app.get("/users/{id}")
async def get_user(
    id: int,
    response: Response,
    if_none_match: str | None = Header(None),
    db = Depends(get_db),
):
    user = await db.get(User, id)
    body = json.dumps(user.dict(), sort_keys=True).encode()
    etag = hashlib.sha256(body).hexdigest()[:16]
    if if_none_match == etag:
        return Response(status_code=304)
    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "private, max-age=60"
    return user

from email.utils import format_datetime

if_modified_since: str | None = Header(None)

if user.updated_at <= parse(if_modified_since):
    return Response(status_code=304)

response.headers["Last-Modified"] = format_datetime(user.updated_at)

import json
from datetime import timedelta

async def get_user_cached(redis, db, user_id: int):
    key = f"user:{user_id}"
    if hit := await redis.get(key):
        return json.loads(hit)
    user = await db.get(User, user_id)
    if user:
        await redis.set(key, json.dumps(user.dict()), ex=300)
    return user.dict() if user else None

# Invalidate on write
async def update_user(redis, db, user_id, **changes):
    user = await db.get(User, user_id)
    for k, v in changes.items(): setattr(user, k, v)
    await db.commit()
    await redis.delete(f"user:{user_id}")

uv add fastapi-cache2 redis

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis.asyncio import Redis

@asynccontextmanager
async def lifespan(app):
    app.state.redis = Redis.from_url("redis://redis")
    FastAPICache.init(RedisBackend(app.state.redis), prefix="myapp")
    yield

@app.get("/posts")
@cache(expire=60)
async def posts():
    return await db.list_posts()

async def get(key, fetch_fn, ex=300):
    if (h := await redis.get(key)) is not None:
        return json.loads(h)
    val = await fetch_fn()
    await redis.set(key, json.dumps(val), ex=ex)
    return val

posts = await get("posts:hot", lambda: db.list_hot_posts())

async def get_swr(key, fetch_fn, fresh_for=60, stale_for=600):
    raw = await redis.get(key)
    meta = await redis.get(f"{key}:ts")
    now = time.time()
    if raw and meta and now - float(meta) < fresh_for:
        return json.loads(raw)
    if raw and meta and now - float(meta) < fresh_for + stale_for:
        # Return stale; refresh in background
        asyncio.create_task(refresh(key, fetch_fn))
        return json.loads(raw)
    val = await fetch_fn()
    await redis.set(key, json.dumps(val))
    await redis.set(f"{key}:ts", str(now))
    return val

async def invalidate(key):
    await redis.delete(key)
    await redis.publish("cache.invalidate", key)

key = f"feed:{user_id}:{page_token}"

async def get_with_lock(key, fetch_fn, ex=300):
    if (v := await redis.get(key)) is not None:
        return json.loads(v)
    lock = f"{key}:lock"
    if await redis.set(lock, "1", nx=True, ex=10):
        try:
            val = await fetch_fn()
            await redis.set(key, json.dumps(val), ex=ex)
            return val
        finally:
            await redis.delete(lock)
    # Someone else is fetching — wait briefly + retry
    await asyncio.sleep(0.1)
    return await get_with_lock(key, fetch_fn, ex=ex)

response.headers["Cache-Control"] = "public, max-age=60, s-maxage=600"
response.headers["CDN-Cache-Control"] = "max-age=600"
response.headers["Cloudflare-CDN-Cache-Control"] = "max-age=600"