FastAPI Cache Decorator Fix
In this tutorial, you'll learn about FastAPI Cache Decorator Fix. We cover key concepts, practical examples, and best practices.
The Problem
Frequently called endpoints hit the database on every request. Adding caching logic to each endpoint manually is repetitive and error-prone.
Quick Fix
Wrong — manual caching in every endpoint
@app.get("/items")
async def get_items():
cached = await redis.get("items:all")
if cached:
return json.loads(cached)
items = await db.fetch_all("SELECT * FROM items")
await redis.setex("items:all", 300, json.dumps([i.dict() for i in items]))
return items
Output: Same 8 lines of caching boilerplate in every endpoint.
Correct — cache decorator
from functools import wraps
from fastapi import Request
import json
def cache(expire: int = 300):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
request = next((a for a in args if isinstance(a, Request)), None)
if not request:
return await func(*args, **kwargs)
key = f"cache:{request.url.path}:{request.query_params}"
cached = await redis.get(key)
if cached:
return json.loads(cached)
response = await func(*args, **kwargs)
await redis.setex(key, expire, json.dumps(response))
return response
return wrapper
return decorator
@app.get("/items")
@cache(expire=60)
async def get_items():
return await db.fetch_all("SELECT * FROM items")
Output: One decorator line replaces 8 lines of caching code.
Cache with cache key function
def cache_key_builder(func, *args, **kwargs):
request = kwargs.get('request') or next(
(a for a in args if isinstance(a, Request)), None
)
if request:
return f"{func.__name__}:{request.url.path}:{request.query_params}"
return f"{func.__name__}:{args}:{kwargs}"
def custom_cache(expire: int = 300, key_builder=cache_key_builder):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
key = key_builder(func, *args, **kwargs)
cached = await redis.get(key)
if cached:
return json.loads(cached)
result = await func(*args, **kwargs)
await redis.setex(key, expire, json.dumps(result))
return result
return wrapper
return decorator
Invalidation decorator
def invalidate_cache(pattern: str):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
result = await func(*args, **kwargs)
# Delete all keys matching pattern
keys = await redis.keys(pattern)
if keys:
await redis.delete(*keys)
return result
return wrapper
return decorator
@app.post("/items")
@invalidate_cache("items:*")
async def create_item(item: Item):
return await db.execute("INSERT INTO items ...")
Per-user cache
def user_cache(expire: int = 300):
def decorator(func):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
user_id = request.state.user.id
key = f"user:{user_id}:{func.__name__}:{request.query_params}"
cached = await redis.get(key)
if cached:
return json.loads(cached)
result = await func(request, *args, **kwargs)
await redis.setex(key, expire, json.dumps(result))
return result
return wrapper
return decorator
Prevention
- Create a reusable cache decorator instead of embedding caching in each endpoint.
- Include query parameters and user ID in cache keys for granularity.
- Implement cache invalidation patterns for write operations.
Common Mistakes with cache decorator
- Forgetting
deriving (Show, Eq)on custom data types needed for debugging - Placing the wildcard pattern first in case expressions, making all subsequent patterns unreachable
- Using
headandtailinstead of pattern matching, causing runtime errors on empty lists
These mistakes appear frequently in real-world FASTAPI code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.
Practice Exercise
Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.
This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro