Cache Invalidation Patterns in FastAPI

Key takeaways:

  • Always set a TTL so a missed invalidation eventually self-corrects.
  • Invalidate in the same code path that mutates the data.
  • Use versioned key prefixes to invalidate a whole group with one increment.
  • Propagate invalidation across instances with events.
  • Test that a write is visible on the next read.

This guide tackles the hard half of Caching Strategies: keeping cached data correct.

The Problem This Solves

Caching is easy until data changes. The failure mode is silent — a cache that keeps serving an old value after an update — and it erodes trust in the API. Disciplined invalidation keeps the cache correct without throwing away its benefit.

Prerequisites

Step-by-Step Implementation

1. TTL as the safety net

# Every cached value expires, so a missed explicit invalidation self-heals.
await redis.set(key, json.dumps(value), ex=300)

2. Write-path invalidation

async def update_product(redis, product_id: int, changes: dict) -> None:
    await write_product(product_id, changes)
    # Same code path that mutates also drops the stale key.
    await redis.delete(f"product:{product_id}")

3. Versioned keys for group invalidation

async def catalog_key(redis, suffix: str) -> str:
    version = await redis.get("catalog:version") or "1"
    return f"catalog:v{version}:{suffix}"


async def invalidate_catalog(redis) -> None:
    # One atomic increment retires every key in the group.
    await redis.incr("catalog:version")

4. Cross-service propagation

# On a write, publish so other instances drop their local copies.
await redis.publish("invalidate", json.dumps({"key": f"product:{product_id}"}))

Edge Cases and Gotchas

  • Race between read and write. A read that repopulates just after a delete can re-cache stale data; a short lock or versioned keys avoids it.
  • Partial keys. Deleting product:1 but not a list that embeds product 1 leaves the list stale; invalidate every key that contains the data.
  • Over-invalidation. Dropping too broadly on every write erases the cache's benefit; scope invalidation to what changed.

Verification

async def test_write_visible_on_next_read(client, redis):
    client.get("/products/1")                    # Populate.
    client.patch("/products/1", json={"name": "new"})
    assert await redis.get("product:1") is None  # Invalidated by the write.
    body = client.get("/products/1").json()
    assert body["name"] == "new"                 # Next read is fresh.