Cache FastAPI responses in Redis

Approach

Write a small cache-aside helper storing JSON in Redis with a TTL. Key by everything that affects the response (path params, query params, and user/tenant if personalized). Invalidate keys (or bump a version prefix) when the underlying data changes, and add HTTP caching headers for public data.

Solution

import json
from collections.abc import Awaitable, Callable
from typing import TypeVar
from fastapi import Response
from redis.asyncio import Redis

redis = Redis.from_url("redis://redis:6379/0", decode_responses=True)
T = TypeVar("T")


async def cached(key: str, ttl: int, compute: Callable[[], Awaitable[T]]) -> T:
    if (hit := await redis.get(key)) is not None:
        return json.loads(hit)
    value = await compute()
    await redis.set(key, json.dumps(value, default=str), ex=ttl)
    return value


@router.get("/categories/{slug}/top-products")
async def top_products(slug: str, db: DB, response: Response):
    version = await redis.get("catalog:version") or "1"
    data = await cached(f"top:{version}:{slug}", 300,
                        lambda: products.top_for_category(db, slug))      # returns JSON-serializable dicts
    response.headers["Cache-Control"] = "public, max-age=60"
    return data


async def on_catalog_changed() -> None:
    await redis.incr("catalog:version")     # invalidates every top:* key at once

Pitfalls

• Caching personalized data under a shared key.
• Caching ORM objects instead of serializable data.
• No TTL — stale data forever if invalidation is missed.

Variations

• fastapi-cache2 provides a @cache decorator with Redis backends.
• Stampede protection with a short Redis lock (SET NX EX).

Further reading: Redis caching docs