FastAPI Redis Caching — Cache-Aside, TTL & Invalidation

🎯 What You Will Build

A FastAPI endpoint that answers in milliseconds by serving repeated reads from Redis instead of hitting the database (or a slow computation) every time — the same cache-aside pattern used in production.

An async Redis connection wired into FastAPI
The cache-aside pattern: check cache → miss → load → store
TTL expiry and sensible key naming
Cache invalidation when data changes
A reusable caching helper you can drop onto any read
The caching traps (stale data, no TTL, caching everything)

Prerequisites: Python 3.11+, a running Redis (docker run -p 6379:6379 redis works), and comfort with Depends() (see fastapi-dependency-injection).

fastapi-cache/
├── .env
├── requirements.txt
└── app/
    ├── __init__.py
    ├── main.py       # app, lifespan, routes
    ├── cache.py      # redis connection + cache helpers
    └── service.py    # the slow data source we're caching

The Mental Model: Cache-Aside

A cache is a fast key-value store that sits beside your slow data source. On every read you ask the cache first. If the value is there (a hit), you return it immediately. If not (a miss), you load from the database, store the result in the cache with an expiry, and return it.

Request  ->  Redis?  --hit-->  return cached value (fast) ✅
              |
             miss
              |
              v
           Database  ->  store in Redis (with TTL)  ->  return value

💡

Ang importante dito: cache the result of expensive work — a heavy query, an aggregation, an external API call. Data that changes every request or is unique per user is usually a poor cache candidate.

The Problem: Repeating Expensive Work (See It First)

Here's an endpoint that recomputes the same expensive result on every call. Pretend get_report is a 500ms aggregation query:

app/service.py

import asyncio


async def get_report(report_id: int) -> dict:
    # Pretend this is a heavy DB aggregation or external API call.
    await asyncio.sleep(0.5)
    return {"report_id": report_id, "total_sales": 128_400, "orders": 512}

app/main.py (no cache — slow every time)

from fastapi import FastAPI

from app.service import get_report

app = FastAPI()


@app.get("/reports/{report_id}")
async def read_report(report_id: int):
    return await get_report(report_id)  # 500ms EVERY request 😩

⚠️

Why this hurts: ten users requesting the same report do the same 500ms work ten times. Under load this crushes your database. The result barely changes — perfect for caching.

Setup + Redis Connection

requirements.txt

fastapi==0.116.1
uvicorn[standard]==0.30.6
redis==5.2.1
pydantic-settings==2.6.1

.env

REDIS_URL=redis://localhost:6379/0
CACHE_TTL_SECONDS=60

We create one Redis client for the whole app and expose it through a dependency. redis.asyncio is the async client shipped with redis-py.

app/cache.py

import redis.asyncio as redis
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    REDIS_URL: str = "redis://localhost:6379/0"
    CACHE_TTL_SECONDS: int = 60
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

# decode_responses=True -> get str back instead of bytes.
redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)


def get_redis() -> redis.Redis:
    return redis_client

app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.cache import redis_client


@asynccontextmanager
async def lifespan(app: FastAPI):
    await redis_client.ping()   # fail fast if Redis is unreachable
    yield
    await redis_client.aclose()  # clean shutdown


app = FastAPI(title="FastAPI Redis Cache", version="0.1.0", lifespan=lifespan)

💡

One client, reused: redis.from_url creates a connection pool. Never open a new client per request — inject the shared one via get_redis.

Implementing Cache-Aside

Now the real thing. Check Redis first; on a miss, load and store with a TTL. Values are JSON strings so any client can read them.

app/main.py

import json
from typing import Annotated

import redis.asyncio as redis
from fastapi import Depends, FastAPI

from app.cache import get_redis, settings
from app.service import get_report

RedisDep = Annotated[redis.Redis, Depends(get_redis)]


@app.get("/reports/{report_id}")
async def read_report(report_id: int, cache: RedisDep):
    key = f"report:{report_id}"

    # 1) try the cache
    cached = await cache.get(key)
    if cached is not None:
        return {"source": "cache", **json.loads(cached)}

    # 2) miss -> do the expensive work
    data = await get_report(report_id)

    # 3) store with TTL so it expires automatically
    await cache.set(key, json.dumps(data), ex=settings.CACHE_TTL_SECONDS)

    return {"source": "db", **data}

✅

First call: source: "db" (~500ms). Every call after (until TTL): source: "cache" (~1ms). Same data, a fraction of the cost.

TTL and Key Design

ex= sets the time-to-live in seconds. When it elapses, Redis deletes the key automatically and the next request repopulates it. TTL is your safety net against stale data.

Data type	Good TTL	Why
Rarely changes (config, categories)	10–60 min	Cheap to serve, low staleness risk
Dashboard / report aggregates	30–120 s	Slightly stale is fine, saves heavy queries
Per-user feed	5–30 s	Freshness matters more
Prices, stock, balances	avoid / very short + invalidate	Stale values cause real bugs

💡

Key naming: use a namespaced, predictable pattern like report:{id} or user:{id}:profile. It makes keys easy to find in redis-cli and easy to invalidate.

⚠️

Always set a TTL. A cache entry with no expiry lives forever — that's how you end up serving data that's hours out of date and slowly filling Redis with garbage keys.

Cache Invalidation on Writes

"There are only two hard things in computer science: cache invalidation and naming things." When the underlying data changes, delete the stale key so the next read repopulates it with fresh data.

app/main.py

from pydantic import BaseModel


class ReportUpdate(BaseModel):
    total_sales: int
    orders: int


@app.put("/reports/{report_id}")
async def update_report(report_id: int, data: ReportUpdate, cache: RedisDep):
    # ... persist the change to your database here ...

    # invalidate: next GET will rebuild the cache from fresh data
    await cache.delete(f"report:{report_id}")

    return {"report_id": report_id, "status": "updated", "cache": "invalidated"}

Need to clear a whole group of keys? Scan by pattern instead of deleting one at a time:

app/cache.py

import redis.asyncio as redis


async def delete_by_prefix(cache: redis.Redis, prefix: str) -> int:
    """Delete all keys matching prefix* (e.g. 'report:'). Safe under load."""
    deleted = 0
    async for key in cache.scan_iter(match=f"{prefix}*", count=100):
        await cache.delete(key)
        deleted += 1
    return deleted

⚠️

Use scan_iter, never KEYS *. KEYS blocks the entire Redis server while it scans — fine on your laptop, a production outage on a big keyspace. scan_iter walks the keyspace in small batches.

A Reusable Cache Helper

Copy-pasting the get/miss/set dance into every endpoint gets old. Wrap it once into a helper that takes a key, a TTL, and a loader function:

app/cache.py

import json
from collections.abc import Awaitable, Callable
from typing import Any

import redis.asyncio as redis


async def cached(
    cache: redis.Redis,
    key: str,
    ttl: int,
    loader: Callable[[], Awaitable[Any]],
) -> Any:
    """Return cached value for key, or run loader(), store it, and return it."""
    hit = await cache.get(key)
    if hit is not None:
        return json.loads(hit)

    value = await loader()
    await cache.set(key, json.dumps(value), ex=ttl)
    return value

app/main.py

from app.cache import cached, settings


@app.get("/reports/{report_id}")
async def read_report(report_id: int, cache: RedisDep):
    data = await cached(
        cache,
        key=f"report:{report_id}",
        ttl=settings.CACHE_TTL_SECONDS,
        loader=lambda: get_report(report_id),
    )
    return data

✅

Now caching is one line. Any read becomes cacheable by wrapping its loader — same key discipline, no repeated boilerplate.

Try It (curl + redis-cli)

1Start Redis and the app

run.sh

docker run -d -p 6379:6379 redis          # start Redis
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
uvicorn app.main:app --reload

2Watch the first (slow) call vs the cached call

# first call: from the DB, ~500ms
$ curl -s -w " (%{time_total}s)\n" http://127.0.0.1:8000/reports/7
{"source":"db","report_id":7,"total_sales":128400,"orders":512} (0.51s)

# second call: from Redis, ~1ms
$ curl -s -w " (%{time_total}s)\n" http://127.0.0.1:8000/reports/7
{"source":"cache","report_id":7,"total_sales":128400,"orders":512} (0.002s)

# inspect the key + its remaining TTL
$ redis-cli GET report:7
"{\"report_id\": 7, \"total_sales\": 128400, \"orders\": 512}"
$ redis-cli TTL report:7
(integer) 54

# update -> key is invalidated -> next GET is "db" again
$ curl -s -X PUT http://127.0.0.1:8000/reports/7 \
    -H "Content-Type: application/json" -d '{"total_sales":200000,"orders":800}'
{"report_id":7,"status":"updated","cache":"invalidated"}

✅

You did it: ~250x faster on cache hits, automatic expiry via TTL, and fresh data after writes thanks to invalidation.

Common Mistakes (and Fixes)

Mistake	Symptom	Fix
Caching without a TTL	Stale data served forever; Redis fills up	Always pass `ex=`
Forgetting to invalidate on writes	Users see old data after an update	`delete` the key on every write
Using `KEYS *` to clear keys	Redis blocks; latency spikes	Use `scan_iter`
Caching per-user or constantly-changing data	Low hit rate, wasted memory	Cache shared, expensive, slow-changing reads
New Redis client per request	Connection exhaustion	One shared client + connection pool
Crashing when Redis is down	Cache outage becomes an app outage	Wrap cache calls; fall back to the DB on error

⚠️

Thundering herd: when a hot key expires, many requests miss at once and all hit the DB together. For very hot keys, use a slightly randomized TTL or a short lock so only one request rebuilds the cache.

💡

Resilience habit: treat the cache as optional. If a Redis call raises, log it and fall back to the data source — a cache being down should slow you down, not take you down.

What's Next

✅

🚀 Recommended next reads:

fastapi-async-sqlalchemy — cache the results of your async DB queries
fastapi-rate-limiting — Redis also powers rate limiting (same client)
fastapi-celery-redis — Redis as a task broker for background jobs
fastapi-error-handling — fall back gracefully when the cache is unavailable

Recap: check the cache first, load on a miss, store with a TTL, and delete on writes. Keep one shared Redis client, name keys predictably, and never let a cache outage become an app outage. That's cache-aside — the workhorse pattern behind most fast APIs.

FastAPI Redis Caching (Cache-Aside, TTL & Invalidation)

🎯 What You Will Build

The Mental Model: Cache-Aside

The Problem: Repeating Expensive Work (See It First)

Setup + Redis Connection

Implementing Cache-Aside

TTL and Key Design

Cache Invalidation on Writes

A Reusable Cache Helper

Try It (curl + redis-cli)

Common Mistakes (and Fixes)

What's Next

About the Author

Thirdy Gayares

🎯 What You Will Build

1The Mental Model: Cache-Aside

2The Problem: Repeating Expensive Work (See It First)

3Setup + Redis Connection

4Implementing Cache-Aside

5TTL and Key Design

6Cache Invalidation on Writes

7A Reusable Cache Helper

8Try It (curl + redis-cli)

9Common Mistakes (and Fixes)

10What's Next

About the Author

Thirdy Gayares

The Mental Model: Cache-Aside

The Problem: Repeating Expensive Work (See It First)

Setup + Redis Connection

Implementing Cache-Aside

TTL and Key Design

Cache Invalidation on Writes

A Reusable Cache Helper

Try It (curl + redis-cli)

Common Mistakes (and Fixes)

What's Next