Redis Response Caching in FastAPI

Key takeaways:

  • Use an async Redis client opened in lifespan so cache calls do not block the loop.
  • Build a stable key from route, path params, and sorted query string.
  • On a hit return cached JSON; on a miss compute, store with a TTL, and return.
  • Include the authenticated principal in the key for user-specific responses.
  • Treat Redis as optional — fall through to the database if it is unavailable.

This guide implements response-level caching from Caching Strategies, and it pairs with serialization performance since a hit skips re-serialization.

The Problem This Solves

A read-heavy endpoint that returns the same large payload to many clients recomputes and re-serializes it every time. Caching the serialized response in Redis turns repeated requests into a single fast memory read, cutting both latency and database load.

Prerequisites

  • An async Redis client (redis.asyncio).
  • A FastAPI app with a lifespan to own the connection.

Step-by-Step Implementation

1. Connect Redis in lifespan

from contextlib import asynccontextmanager

from fastapi import FastAPI
from redis.asyncio import Redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = Redis.from_url(app.state.settings.redis_url, decode_responses=True)
    yield
    await app.state.redis.aclose()

2. Build a stable key and cache-aside helper

import json
from urllib.parse import urlencode

from fastapi import Request
from redis.asyncio import Redis


def cache_key(request: Request) -> str:
    # Sorted query string → equivalent queries share one key.
    qs = urlencode(sorted(request.query_params.multi_items()))
    return f"resp:{request.url.path}?{qs}"


async def cached_json(redis: Redis, key: str, loader, ttl: int = 120):
    try:
        if hit := await redis.get(key):
            return json.loads(hit)               # Hit: skip compute + serialize.
    except Exception:
        pass                                     # Cache optional: fall through.
    value = await loader()
    try:
        await redis.set(key, json.dumps(value), ex=ttl)
    except Exception:
        pass
    return value

3. Use it in a route

@router.get("/catalog")
async def catalog(request: Request):
    redis = request.app.state.redis
    return await cached_json(redis, cache_key(request), load_catalog, ttl=300)

4. Invalidate on writes

@router.post("/catalog/items")
async def add_item(item: ItemIn, request: Request):
    await save_item(item)
    # Drop the cached listing so the next read reflects the new item.
    await request.app.state.redis.delete("resp:/catalog?")

Edge Cases and Gotchas

  • User-specific data. Include the principal in the key, or you will serve one user's data to another.
  • Large values. Cap cached payload size; very large values pressure Redis memory.
  • Thundering herd. A hot key expiring under load can stampede; add a short lock, per Caching Strategies.

Verification

async def test_second_request_is_cached(client, db_spy):
    client.get("/catalog")        # Miss → loads.
    db_spy.reset()
    client.get("/catalog")        # Hit → no DB load.
    assert db_spy.load_count == 0