FastAPI中实现用户级缓存需将user.id等唯一标识纳入缓存键:①用lru_cache装饰内部函数并传user.id;②用contextvars实现单请求隔离;③生产环境推荐Redis+用户命名空间+TTL。

缓存依赖函数但按用户隔离的关键点
FastAPI 本身不提供带用户上下文的缓存机制,cache 类装饰器(比如 functools.lru_cache)是全局共享的,无法区分 current_user。必须手动把用户标识(如 user.id 或 user.token)纳入缓存键,且确保依赖函数能访问到当前用户。
用 lru_cache 手动构造用户级缓存键
不能直接装饰依赖函数,因为 lru_cache 在模块加载时就绑定,而 current_user 是运行时注入的。正确做法是:在依赖函数内部调用一个带参数的缓存函数,并把 user.id 作为显式参数传入。
示例:
from functools import lru_cache from fastapi import Depends, HTTPException@lru_cache(maxsize=128) def _get_user_prefs_cached(user_id: int) -> dict:
模拟耗时操作,如查 DB 或远程调用
return {"theme": "dark", "lang": "zh"}def get_user_prefs(current_user = Depends(get_current_user)): if not current_user: raise HTTPException(401) return _get_user_prefs_cached(current_user.id)
-
_get_user_prefs_cached是真正被缓存的函数,参数必须是可哈希的(int、str等),不能传user对象本身 - 依赖函数
get_user_prefs不加缓存装饰,只负责提取user.id并转发 - 注意
maxsize要合理,避免内存堆积;若用户量大,考虑用redis替代lru_cache
用 contextvars + 自定义缓存容器(适合更复杂场景)
当需要动态控制缓存生命周期(例如请求结束自动清理)、或缓存值依赖多个上下文变量(如 user.id + request.headers["X-Region"])时,lru_cache 就不够用了。可用 contextvars.ContextVar 绑定每请求独立的缓存字典。
示例:
import contextvars from typing import Dict, Any_user_cache_var = contextvars.ContextVar("user_cache", default={})
def get_cached_user_data(key: str, factory_func, *args, *kwargs) -> Any: cache = _user_cache_var.get() if key not in cache: cache[key] = factory_func(args, **kwargs) _user_cache_var.set(cache) return cache[key]
在依赖中使用:
def get_user_report(current_user = Depends(get_current_user)): return get_cached_userdata( f"report{current_user.id}", generate_report_for_user, current_user.id )
- 每个请求有独立的
cache字典,天然按用户/请求隔离 - 需配合中间件在请求开始前初始化
_user_cache_var,否则可能复用上一个请求的缓存 - 不适用于跨请求复用(比如同一用户多次请求想复用),它只在单次请求内有效
Redis 缓存 + 用户 ID 命名空间(生产推荐)
真实项目中,lru_cache 和 contextvars 都受限于进程内存和生命周期。用 Redis 可实现跨进程、带 TTL、支持剔除策略的用户级缓存。
关键点:
- 缓存键必须包含用户唯一标识,例如
f"user:{user.id}:prefs"或f"dep:user_prefs:{user.id}" - 避免硬编码前缀,建议封装成函数:
make_cache_key("user_prefs", user.id) - 务必设
ex(TTL),防止脏数据长期滞留;对敏感数据,删除时机要明确(如用户登出时清 Redis 键) - 如果依赖函数抛异常,别让错误结果也被缓存——加
try/except控制是否写入缓存
FastAPI 依赖中调用 Redis 示例(使用 aioredis):
async def get_user_settings(
current_user = Depends(get_current_user),
redis = Depends(get_redis_client)
):
key = f"user:{current_user.id}:settings"
cached = await redis.get(key)
if cached:
return json.loads(cached)
data = await fetch_from_db(current_user.id) # 实际逻辑
await redis.set(key, json.dumps(data), ex=300) # 5 分钟过期
return data
用户隔离不是加个 @cache 就能解决的事。核心永远是:缓存键里有没有稳定、唯一、可预测的用户标识,以及这个标识能不能在缓存读写时被可靠拿到。漏掉任意一环,就会出现 A 用户看到 B 用户的缓存结果。










