FastAPI路由Redis缓存装饰器：如何获取路由URL与请求参数

核心思路 #

FastAPI会为被路由装饰器修饰的函数添加__fastapi_route__属性，通过它可直接获取路由路径；查询参数会被FastAPI注入到函数的关键字参数（kwargs）中，直接读取即可。结合异步Redis客户端实现缓存，无需大幅修改原有路由函数。

完整实现代码 #

先安装异步Redis客户端：

pip install redis

装饰器与路由代码：

from functools import wraps
from fastapi import APIRouter
import redis.asyncio as redis
import json

# 初始化异步Redis客户端
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
router = APIRouter()

def cache(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 从函数属性中获取路由路径
        route_path = func.__fastapi_route__.path
        # 直接从关键字参数中取出查询参数q
        q_param = kwargs.get('q')
        
        # 无q参数时直接执行原函数
        if not q_param:
            return await func(*args, **kwargs)
        
        # 构建唯一缓存键，避免不同请求缓存冲突
        cache_key = f"fastapi_cache:{route_path}:{q_param}"
        # 尝试读取Redis缓存
        cached_result = await redis_client.get(cache_key)
        
        if cached_result:
            # 缓存存在时反序列化并返回
            return json.loads(cached_result)
        
        # 缓存不存在时执行原函数
        result = await func(*args, **kwargs)
        # 将结果存入Redis，设置60秒过期时间
        await redis_client.setex(cache_key, 60, json.dumps(result))
        
        return result
    return wrapper

# 注意装饰器顺序：@cache要放在@router.get之后
@router.get("/query/", status_code=200)
@cache
async def query(q: str):
    # 模拟从数据库获取数据的逻辑
    res = {"query": q, "data": f"对应查询结果：{q}"}
    return res

关键注意事项 #

• 装饰器顺序：@cache必须放在@router.get之后，FastAPI完成路由注册后，函数才会生成__fastapi_route__属性，才能正确读取路由路径。
• 异步适配：原路由函数是异步的，装饰器的wrapper必须定义为async，同时使用异步Redis客户端，避免阻塞FastAPI的事件循环。
• 缓存键设计：用「路由路径+查询参数」组合作为缓存键，确保不同路径、不同参数的请求缓存独立。
• 结果序列化：Redis仅支持字符串存储，需将返回的字典/对象序列化为JSON字符串，取出时再反序列化。

内容的提问来源于stack exchange，提问作者Anonyo Noor