FastAPI Caching with Redis

This skill provides production-ready caching patterns using Redis for FastAPI applications.

Redis Client Setup

Connection Configuration

# app/infrastructure/cache.py
import redis.asyncio as redis
from typing import Optional, Any
import json
from app.config import get_settings

settings = get_settings()

class RedisCache:
    def __init__(self):
        self._pool: Optional[redis.ConnectionPool] = None
        self._client: Optional[redis.Redis] = None

    async def connect(self):
        self._pool = redis.ConnectionPool.from_url(
            settings.redis_url,
            encoding="utf-8",
            decode_responses=True,
            max_connections=20
        )
        self._client = redis.Redis(connection_pool=self._pool)

    async def disconnect(self):
        if self._pool:
            await self._pool.disconnect()

    @property
    def client(self) -> redis.Redis:
        if not self._client:
            raise RuntimeError("Redis not connected")
        return self._client

    async def get(self, key: str) -> Optional[Any]:
        value = await self.client.get(key)
        if value:
            return json.loads(value)
        return None

    async def set(
        self,
        key: str,
        value: Any,
        expire: int = 3600
    ):
        await self.client.setex(
            key,
            expire,
            json.dumps(value, default=str)
        )

    async def delete(self, key: str):
        await self.client.delete(key)

    async def delete_pattern(self, pattern: str):
        """Delete all keys matching pattern."""
        keys = []
        async for key in self.client.scan_iter(match=pattern):
            keys.append(key)
        if keys:
            await self.client.delete(*keys)

    async def exists(self, key: str) -> bool:
        return await self.client.exists(key) > 0

cache = RedisCache()

Lifespan Integration

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    await cache.connect()
    yield
    await cache.disconnect()

app = FastAPI(lifespan=lifespan)

Cache Decorator

# app/core/cache.py
from functools import wraps
from typing import Callable, Optional
import hashlib
import json

def cached(
    prefix: str,
    expire: int = 3600,
    key_builder: Optional[Callable] = None
):
    """
    Cache decorator for async functions.

    Args:
        prefix: Cache key prefix
        expire: TTL in seconds
        key_builder: Custom function to build cache key
    """
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Build cache key
            if key_builder:
                cache_key = f"{prefix}:{key_builder(*args, **kwargs)}"
            else:
                # Default: hash all arguments
                key_data = json.dumps(
                    {"args": args[1:], "kwargs": kwargs},
                    sort_keys=True,
                    default=str
                )
                key_hash = hashlib.md5(key_data.encode()).hexdigest()
                cache_key = f"{prefix}:{key_hash}"

            # Try cache
            cached_value = await cache.get(cache_key)
            if cached_value is not None:
                return cached_value

            # Execute and cache
            result = await func(*args, **kwargs)
            await cache.set(cache_key, result, expire)

            return result
        return wrapper
    return decorator

Usage Examples

class UserService:
    @cached(prefix="user", expire=300)
    async def get_by_id(self, user_id: str) -> Optional[dict]:
        user = await User.get(user_id)
        return user.model_dump() if user else None

    @cached(
        prefix="users_list",
        expire=60,
        key_builder=lambda self, skip, limit, **kw: f"{skip}:{limit}"
    )
    async def get_all(self, skip: int = 0, limit: int = 100) -> list:
        users = await User.find_all().skip(skip).limit(limit).to_list()
        return [u.model_dump() for u in users]

    async def create(self, data: UserCreate) -> User:
        user = await User(**data.model_dump()).insert()
        # Invalidate list cache
        await cache.delete_pattern("users_list:*")
        return user

    async def update(self, user_id: str, data: UserUpdate) -> User:
        user = await self.get_by_id_uncached(user_id)
        await user.set(data.model_dump(exclude_unset=True))
        # Invalidate caches
        await cache.delete(f"user:{user_id}")
        await cache.delete_pattern("users_list:*")
        return user

Response Caching Middleware

# app/middleware/cache.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import hashlib

class ResponseCacheMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, cache_ttl: int = 60, cache_methods: list = None):
        super().__init__(app)
        self.cache_ttl = cache_ttl
        self.cache_methods = cache_methods or ["GET"]

    async def dispatch(self, request: Request, call_next):
        # Only cache specified methods
        if request.method not in self.cache_methods:
            return await call_next(request)

        # Skip if cache-control: no-cache
        if request.headers.get("cache-control") == "no-cache":
            return await call_next(request)

        # Build cache key
        cache_key = self._build_key(request)

        # Try cache
        cached = await cache.get(cache_key)
        if cached:
            return Response(
                content=cached["body"],
                status_code=cached["status"],
                headers={**cached["headers"], "X-Cache": "HIT"}
            )

        # Execute request
        response = await call_next(request)

        # Cache successful responses
        if 200 <= response.status_code < 300:
            body = b""
            async for chunk in response.body_iterator:
                body += chunk

            await cache.set(cache_key, {
                "body": body.decode(),
                "status": response.status_code,
                "headers": dict(response.headers)
            }, self.cache_ttl)

            return Response(
                content=body,
                status_code=response.status_code,
                headers={**response.headers, "X-Cache": "MISS"}
            )

        return response

    def _build_key(self, request: Request) -> str:
        key_data = f"{request.method}:{request.url.path}:{request.query_params}"
        return f"response:{hashlib.md5(key_data.encode()).hexdigest()}"

Cache-Aside Pattern

class CachedUserRepository:
    def __init__(self, cache: RedisCache, db: Database):
        self.cache = cache
        self.db = db

    async def get(self, user_id: str) -> Optional[User]:
        # 1. Check cache
        cache_key = f"user:{user_id}"
        cached = await self.cache.get(cache_key)
        if cached:
            return User(**cached)

        # 2. Query database
        user = await self.db.users.find_one({"_id": user_id})
        if not user:
            return None

        # 3. Store in cache
        await self.cache.set(cache_key, user.model_dump(), expire=300)

        return user

    async def save(self, user: User) -> User:
        # 1. Save to database
        await self.db.users.update_one(
            {"_id": user.id},
            {"$set": user.model_dump()},
            upsert=True
        )

        # 2. Invalidate cache
        await self.cache.delete(f"user:{user.id}")

        return user

Write-Through Pattern

class WriteThroughUserRepository:
    async def save(self, user: User) -> User:
        # 1. Write to database
        await self.db.users.update_one(
            {"_id": user.id},
            {"$set": user.model_dump()},
            upsert=True
        )

        # 2. Update cache immediately
        await self.cache.set(
            f"user:{user.id}",
            user.model_dump(),
            expire=300
        )

        return user

Cache Invalidation Patterns

# Event-based invalidation
class CacheInvalidator:
    def __init__(self, cache: RedisCache):
        self.cache = cache

    async def on_user_updated(self, user_id: str):
        await self.cache.delete(f"user:{user_id}")
        await self.cache.delete_pattern("users_list:*")
        await self.cache.delete_pattern(f"user_orders:{user_id}:*")

    async def on_product_updated(self, product_id: str):
        await self.cache.delete(f"product:{product_id}")
        await self.cache.delete_pattern("products_list:*")
        await self.cache.delete_pattern("category_products:*")

Additional Resources

Reference Files

For detailed patterns:

• references/patterns.md

  - Cache patterns (write-behind, read-through)

• references/distributed.md

  - Distributed caching, cache stampede

• references/monitoring.md

  - Cache hit rates, memory usage

Example Files

Working examples in

examples/

:

• examples/cache_service.py

  - Complete cache service

• examples/cached_repository.py

  - Repository with caching

• examples/cache_middleware.py

  - Response caching middleware