Caching Strategies in FastAPI

Caching is trading freshness for speed: storing hot, rarely-changing data in a fast store such as Redis so most requests skip the database entirely, in exchange for managing staleness and invalidation.

This topic is part of Async, Background Tasks and Observability. It relieves the async database under read-heavy load and complements serialization performance by avoiding repeated work for the same data.

The cache-aside read path A request checks the cache. On a hit it returns immediately. On a miss it reads the database, writes the value into the cache with a TTL, and returns it. Request Cache (Redis) check key hit → return miss → DB then populate set TTL
Cache-aside: a hit returns immediately, a miss falls through to the database and populates the cache with a TTL for the next reader.

Core Mechanics: Cache-Aside with Redis

The application owns the cache logic: check first, fall through on a miss, populate, return. An async Redis client keeps the cache calls off the event loop just like the database.

import json

from redis.asyncio import Redis


async def get_product(redis: Redis, product_id: int) -> dict:
    key = f"product:{product_id}"
    if cached := await redis.get(key):
        return json.loads(cached)              # Hit: skip the database entirely.
    product = await load_product_from_db(product_id)
    # Populate with a TTL so the value refreshes after it goes stale.
    await redis.set(key, json.dumps(product), ex=300)
    return product

Production Implementation: Invalidation and Stampede Protection

Invalidation is the hard half. Delete or update the key when the underlying data changes, and protect popular keys from stampedes with a short lock so only one request recomputes.

async def update_product(redis: Redis, product_id: int, changes: dict) -> None:
    await write_product_to_db(product_id, changes)
    # Explicit invalidation: drop the key so the next read repopulates fresh.
    await redis.delete(f"product:{product_id}")


async def get_with_lock(redis: Redis, key: str, loader) -> dict:
    if cached := await redis.get(key):
        return json.loads(cached)
    # Only one caller computes; others briefly retry, preventing a stampede.
    if await redis.set(f"lock:{key}", "1", nx=True, ex=5):
        value = await loader()
        await redis.set(key, json.dumps(value), ex=300)
        await redis.delete(f"lock:{key}")
        return value
    await asyncio.sleep(0.05)
    return await get_with_lock(redis, key, loader)

The Redis response-caching build-out is detailed in Redis Response Caching in FastAPI.

Async and Performance Notes

Use an async Redis client so cache access yields the loop. Cache the representation that removes the most repeated work — a serialized response skips re-serialization for large nested payloads, while a cached model is more reusable across endpoints. Keep values small and set a TTL on every key so a forgotten invalidation cannot serve stale data forever.

Testing Strategy

Assert hit, miss, and invalidation behavior with a fake or test Redis:

async def test_cache_aside(redis, db):
    first = await get_product(redis, 1)       # Miss: loads from DB, populates.
    db.calls.clear()
    second = await get_product(redis, 1)      # Hit: no DB call.
    assert second == first and not db.calls
    await update_product(redis, 1, {"name": "new"})
    assert await redis.get("product:1") is None   # Invalidated.

Failure Modes and Debugging

  • No TTL. A key with no expiry serves stale data indefinitely after a missed invalidation; always set one.
  • Stampedes. A hot key expiring under load floods the database; add a lock or pre-refresh.
  • Caching user-specific data globally. A shared key leaking per-user data is a security bug; include the identity in the key.
  • Sync Redis client. Blocks the loop; use the async client, per Async Correctness and Concurrency.