fastapi 如何让依赖函数支持缓存但按用户隔离

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

缓存依赖函数但按用户隔离的关

键点

FastAPI 本身不提供带用户上下文的缓存机制,cache 类装饰器(比如 functools.lru_cache)是全局共享的,无法区分 current_user。必须手动把用户标识(如 user.iduser.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 是真正被缓存的函数,参数必须是可哈希的(intstr 等),不能传 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_cachecontextvars 都受限于进程内存和生命周期。用 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 用户的缓存结果。