FastAPI Redis Caching

Redis Caching in FastAPI

pip install redis

from redis import Redis
from functools import wraps
import json

redis_client = Redis(host="localhost", port=6379, decode_responses=True)

def cache(ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key    = func.__name__ + ":" + str(args)
            cached = redis_client.get(key)
            if cached: return json.loads(cached)
            result = await func(*args, **kwargs)
            redis_client.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@app.get("/posts")
@cache(ttl=120)
async def list_posts(db: Session = Depends(get_db)):
    return crud.get_posts(db)

def bust_cache(prefix: str):
    for key in redis_client.scan_iter(prefix + ":*"):
        redis_client.delete(key)