Caching Strategies for FastAPI: Redis, In-Memory, and HTTP Cache Headers

Caching is one of those things that seems straightforward until you realize there are three different layers to think about. I’ve seen teams waste time implementing Redis caching for data that could be cached in-memory, or forget HTTP cache headers and lose optimization opportunities on the client side.

This guide covers the practical strategies: when to use each layer, how to implement them, and real performance comparisons.

The Three Caching Layers

• HTTP cache headers (browser/CDN cache) — Free, no server hit
• In-memory cache (app instance) — Fast but not shared between instances
• Redis cache (distributed) — Persists across instances, shared state

Most production apps need all three. They complement each other, not replace each other.

Layer 1: HTTP Cache Headers (The Overlooked One)

This is the cheapest optimization. Let browsers and CDNs cache your responses so they never hit your server.

Basic Cache-Control:

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(product_id: int, response: Response):
    """Product data that changes infrequently"""
    product = {
        "id": product_id,
        "name": "Widget",
        "price": 99.99
    }

    # Cache in browser for 1 hour
    response.headers["Cache-Control"] = "public, max-age=3600"

    return product

Common Cache-Control values:

• max-age=3600 - Cache for 3600 seconds (1 hour)
• public - Any cache can store this (browser, CDN, proxy)
• private - Only browser cache, not CDN
• no-cache - Must revalidate with server before using
• no-store - Never cache this response

ETag for Smart Revalidation

ETags let browsers reuse cached data without re-downloading. If data hasn’t changed, server returns 304 Not Modified with no body.

from fastapi import FastAPI, Response, Request
from fastapi.responses import Response as FastAPIResponse
import hashlib
import json

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request):
    """User data with ETag support"""
    user_data = {
        "id": user_id,
        "name": "Alice",
        "email": "alice@example.com"
    }

    # Generate ETag from data
    data_str = json.dumps(user_data, sort_keys=True)
    etag = f'"{hashlib.md5(data_str.encode()).hexdigest()}"'

    # Check if client has current version
    client_etag = request.headers.get("if-none-match")
    if client_etag == etag:
        # Data hasn't changed, return 304
        return FastAPIResponse(
            status_code=304,
            headers={"ETag": etag}
        )

    # Data changed or no ETag, send full response
    return FastAPIResponse(
        content=json.dumps(user_data),
        media_type="application/json",
        headers={
            "ETag": etag,
            "Cache-Control": "public, max-age=300"
        }
    )

How it works:

• Server sends ETag hash with response
• The browser caches the response
• Next request, the browser sends If-None-Match: <etag>
• Server compares ETags. If the same, return 304
• The browser uses the cached copy

When to use HTTP caching:

• Public data (product catalogs, blog posts)
• Data that doesn’t change often
• Static or semi-static content
• Any GET endpoint that returns the same data for a period

Layer 2: In-Memory Cache

Great for single-instance apps or data that’s expensive to compute but safe to lose on restart.

Using functools.lru_cache

from functools import lru_cache
from fastapi import FastAPI
import time

app = FastAPI()

@lru_cache(maxsize=128)
def get_user_permissions(user_id: int) -> list[str]:
    """Compute permissions - cached automatically"""
    time.sleep(0.5)  # Simulate expensive check
    # In real app: check database, compute from rules, etc.
    return ["read", "write", "delete"]

@app.get("/users/{user_id}/permissions")
async def get_permissions(user_id: int):
    """Endpoint using cached function"""
    permissions = get_user_permissions(user_id)
    return {"user_id": user_id, "permissions": permissions}

@app.post("/users/{user_id}/permissions/clear")
async def clear_permissions_cache(user_id: int):
    """Invalidate cache for specific user"""
    get_user_permissions.cache_clear()
    return {"message": "Cache cleared"}

Note: lru_cache only works with regular functions, not async def. Keep cached functions synchronous.

Using cachetools for TTL

# pip install cachetools

from fastapi import FastAPI
from cachetools import TTLCache
import time

app = FastAPI()

# Cache 1000 items, each lives 5 minutes
cache = TTLCache(maxsize=1000, ttl=300)

def expensive_computation(user_id: int) -> dict:
    """Simulate expensive operation"""
    time.sleep(1)
    return {"user_id": user_id, "result": "expensive data"}

@app.get("/data/{user_id}")
async def get_data(user_id: int):
    """Endpoint with in-memory TTL cache"""
    cache_key = f"user_data_{user_id}"

    # Check cache
    if cache_key in cache:
        return {"cached": True, "data": cache[cache_key]}

    # Compute and cache
    result = expensive_computation(user_id)
    cache[cache_key] = result

    return {"cached": False, "data": result}

Performance: ~0.5ms for cache hit vs 1000ms for full computation.

When to use in-memory cache:

• Single instance deployments
• Config data, permissions, lookup tables
• Results of expensive computations
• Data safe to lose on restart
• Total cache size under 100MB

Limitations:

• Not shared between app instances
• Lost on restart
• Uses app memory

Layer 3: Redis Cache

When you need cache shared across instances or that survives restarts, use Redis.

# pip install redis

from fastapi import FastAPI
from contextlib import asynccontextmanager
import redis.asyncio as redis
import json

# Global redis client
redis_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage Redis connection lifecycle"""
    global redis_client

    # Startup: connect to Redis
    redis_client = redis.Redis(
        host="localhost",
        port=6379,
        decode_responses=True
    )

    try:
        await redis_client.ping()
        print("✓ Connected to Redis")
    except redis.ConnectionError:
        print("✗ Redis connection failed")
        redis_client = None

    yield

    # Shutdown: close Redis connection
    if redis_client:
        await redis_client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    """Post data cached in Redis"""
    cache_key = f"post:{post_id}"

    # Try Redis first
    if redis_client:
        try:
            cached = await redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
        except redis.RedisError as e:
            print(f"Redis error: {e}")
            # Fall through to database

    # Fetch from database
    post = {
        "id": post_id,
        "title": "Post Title",
        "content": "Post content here..."
    }

    # Cache for 1 hour
    if redis_client:
        try:
            await redis_client.setex(
                cache_key,
                3600,
                json.dumps(post)
            )
        except redis.RedisError as e:
            print(f"Redis cache write failed: {e}")

    return post

Reusable Cache Helper

from typing import Callable, Any
import json

async def cached(
    key: str,
    ttl: int,
    fetch_func: Callable[[], Any]
) -> Any:
    """
    Generic cache helper with fallback

    Args:
        key: Redis key
        ttl: Time to live in seconds
        fetch_func: Function to call if cache misses
    """
    # Try cache
    if redis_client:
        try:
            cached = await redis_client.get(key)
            if cached:
                return json.loads(cached)
        except redis.RedisError:
            pass  # Fall through to fetch

    # Fetch data
    data = await fetch_func() if callable(fetch_func) else fetch_func

    # Store in cache
    if redis_client:
        try:
            await redis_client.setex(key, ttl, json.dumps(data))
        except redis.RedisError:
            pass  # Cache write failed, but we have data

    return data

# Usage
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """User endpoint with helper"""

    async def fetch_user():
        # Simulate database call
        return {"id": user_id, "name": "Alice"}

    user = await cached(
        key=f"user:{user_id}",
        ttl=1800,  # 30 minutes
        fetch_func=fetch_user
    )

    return user

Cache Invalidation

The hard part of caching is keeping it fresh.

@app.put("/users/{user_id}")
async def update_user(user_id: int, name: str):
    """Update user and invalidate cache"""

    # Update database
    updated_user = {"id": user_id, "name": name}
    # In real app: await db.update_user(user_id, name)

    # Invalidate cache
    if redis_client:
        try:
            await redis_client.delete(f"user:{user_id}")
        except redis.RedisError as e:
            print(f"Cache invalidation failed: {e}")

    return updated_user

@app.delete("/cache/pattern/{pattern}")
async def clear_cache_pattern(pattern: str):
    """
    Clear all keys matching pattern
    Example: /cache/pattern/user:* clears all user caches
    """
    if not redis_client:
        return {"error": "Redis not available"}

    try:
        keys = []
        async for key in redis_client.scan_iter(match=pattern):
            keys.append(key)

        if keys:
            await redis_client.delete(*keys)

        return {"cleared": len(keys), "pattern": pattern}
    except redis.RedisError as e:
        return {"error": str(e)}

Warning: redis_client.flushdb() clears the entire Redis database. Use only in development.

When to use Redis:

• Multi-instance deployments
• Shared state across services
• Cache that survives restarts
• Session data
• Rate limiting data
• Cache larger than available RAM per instance

Combined Strategy: The Production Approach

Use all three layers together:

from fastapi import FastAPI, Response, Request
from fastapi.responses import Response as FastAPIResponse
from contextlib import asynccontextmanager
import redis.asyncio as redis
import json
import hashlib
from functools import lru_cache

redis_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    redis_client = redis.Redis(host="localhost", decode_responses=True)

    try:
        await redis_client.ping()
    except redis.ConnectionError:
        redis_client = None

    yield

    if redis_client:
        await redis_client.aclose()

app = FastAPI(lifespan=lifespan)

# Layer 1: In-memory for config (never changes during runtime)
@lru_cache(maxsize=1)
def get_app_config() -> dict:
    """App config - cached in memory"""
    return {
        "feature_flags": {"new_ui": True},
        "api_version": "1.0"
    }

async def fetch_product_from_db(product_id: int) -> dict:
    """Simulate database fetch"""
    # In real app: await db.products.find_one({"id": product_id})
    return {
        "id": product_id,
        "name": f"Product {product_id}",
        "price": 99.99
    }

@app.get("/products/{product_id}")
async def get_product(product_id: int, request: Request):
    """Product endpoint with 3-layer caching"""

    # Layer 2: Check Redis
    redis_key = f"product:{product_id}"
    data = None

    if redis_client:
        try:
            cached = await redis_client.get(redis_key)
            if cached:
                data = json.loads(cached)
        except redis.RedisError:
            pass

    # Layer 3: Fetch from database if not cached
    if not data:
        data = await fetch_product_from_db(product_id)

        # Store in Redis for 1 hour
        if redis_client:
            try:
                await redis_client.setex(redis_key, 3600, json.dumps(data))
            except redis.RedisError:
                pass

    # Layer 1: HTTP cache headers + ETag
    data_str = json.dumps(data, sort_keys=True)
    etag = f'"{hashlib.md5(data_str.encode()).hexdigest()}"'

    # Check if client has current version
    if request.headers.get("if-none-match") == etag:
        return FastAPIResponse(
            status_code=304,
            headers={"ETag": etag}
        )

    # Return with cache headers
    return FastAPIResponse(
        content=data_str,
        media_type="application/json",
        headers={
            "ETag": etag,
            "Cache-Control": "public, max-age=1800"  # 30 min browser cache
        }
    )

@app.get("/config")
async def get_config(response: Response):
    """Config cached in memory only"""
    config = get_app_config()
    response.headers["Cache-Control"] = "public, max-age=86400"  # 24 hours
    return config

Remember to Don’t cache:

• Real-time data (stock prices, sensor data)
• User-specific sensitive data (without careful expiration)
• Large responses you can optimize differently

Common Mistakes

Mistake 1: No TTL on Redis keys

# Bad: Cache never expires
await redis_client.set("user_profile", user_data)

# Good: Always set TTL
await redis_client.setex("user_profile", 3600, user_data)

Mistake 2: Forgetting to invalidate

# Bad: Stale cache after update
@app.put("/users/{user_id}")
async def update_user(user_id: int, name: str):
    await db.update_user(user_id, name)
    return {"updated": True}

# Good: Invalidate on write
@app.put("/users/{user_id}")
async def update_user(user_id: int, name: str):
    await db.update_user(user_id, name)
    await redis_client.delete(f"user:{user_id}")
    return {"updated": True}

Mistake 3: In-memory cache in multi-instance setup

# Bad: Each instance has different cache
cache = {}  # Instance A and B have separate caches

# Good: Use Redis for shared cache
await redis_client.get(key)  # All instances see same cache

Mistake 4: No error handling for Redis

# Bad: App crashes if Redis is down
cached = await redis_client.get(key)

# Good: Fallback to database
try:
    cached = await redis_client.get(key)
except redis.RedisError:
    # Redis down, fetch from database
    cached = None

Mistake 5: Cache stampede

When cache expires, multiple requests hit database simultaneously.

# Solution: Use Redis lock
import asyncio

async def get_with_lock(key: str, ttl: int, fetch_func):
    """Prevent cache stampede with lock"""

    # Try cache first
    cached = await redis_client.get(key)
    if cached:
        return json.loads(cached)

    # Try to acquire lock
    lock_key = f"{key}:lock"
    lock_acquired = await redis_client.set(
        lock_key,
        "1",
        ex=10,  # Lock expires in 10 seconds
        nx=True  # Only set if not exists
    )

    if lock_acquired:
        # We got the lock, fetch data
        try:
            data = await fetch_func()
            await redis_client.setex(key, ttl, json.dumps(data))
            return data
        finally:
            await redis_client.delete(lock_key)
    else:
        # Someone else is fetching, wait a bit and retry
        await asyncio.sleep(0.1)
        return await get_with_lock(key, ttl, fetch_func)

Caching is powerful but adds complexity. Start with the simplest layer that solves your problem, then add more layers as needed.

If you found this guide useful, consider following for more insights on practical Python and FastAPI development.