如何在FastAPI应用中实现函数级Cache-Control策略的传播?
基于FastAPI实现函数级别的Cache-Control缓存控制
我们在使用FastAPI和FastAPICache时,通常可以通过Cache-Control请求头控制API响应是强制重新计算结果,还是允许从缓存返回。现在需要把这种缓存控制逻辑下沉到函数级别——比如端点foo调用的工具函数bar(仅接收参数a,可能位于独立模块),不需要显式传递Cache-Control参数,而是能自动从当前请求上下文获取缓存策略并执行对应的逻辑。
实现方案
核心思路是利用FastAPI的依赖注入结合Python的contextvars,将请求的Cache-Control信息绑定到当前请求的上下文,让下游函数可以直接获取。
1. 定义上下文变量
创建一个通用模块(比如common.py),用ContextVar存储当前请求的Cache-Control信息:
from contextvars import ContextVar from starlette.datastructures import CacheControl # 上下文变量,默认值为None,隔离不同请求的缓存控制信息 cache_control_ctx: ContextVar[CacheControl | None] = ContextVar("cache_control", default=None)
2. 注入请求上下文的Cache-Control
在主路由模块(main.py)中,定义一个依赖函数,解析请求头中的Cache-Control并写入上下文:
from fastapi import APIRouter, Request, Depends from starlette.datastructures import CacheControl from common import cache_control_ctx from utils import bar router = APIRouter() def inject_cache_control(request: Request): # 解析请求头的Cache-Control,无该头时传入空字符串 cache_control = CacheControl(request.headers.get("Cache-Control", "")) # 将解析后的对象存入上下文 cache_control_ctx.set(cache_control) return cache_control @router.post("/foo") def foo(request: Request, _: CacheControl = Depends(inject_cache_control)): # 此处假设request.body.a是已解析的请求体参数,实际需根据你的请求模型调整 return bar(a=request.body.a)
3. 函数级别缓存控制逻辑
在工具模块(utils.py)中,bar函数直接从上下文获取Cache-Control,执行对应的缓存逻辑:
from common import cache_control_ctx # 假设你已初始化缓存实例cache,以及定义了耗时计算函数some_expensive_computation def bar(a: int): cache_control = cache_control_ctx.get() # 无上下文或要求no_cache时,强制重新计算 if not cache_control or cache_control.no_cache: val = some_expensive_computation(a) else: # 优先从缓存获取,不存在则计算 val = cache.get(a, default=lambda: some_expensive_computation(a)) # 若未要求no_store,将结果存入缓存 if cache_control and not cache_control.no_store: cache.set(a, val) return val
关键说明
ContextVar是请求隔离的,多并发请求下不会出现上下文混乱的问题。- 如果
bar函数需要在非请求场景调用(比如脚本、单元测试),可以给默认逻辑(比如强制计算),避免因上下文为空导致异常。 - 使用Starlette自带的
CacheControl类可以直接解析no_cache、no_store等指令,无需自己处理复杂的HTTP头字符串。
内容的提问来源于stack exchange,提问作者angryweasel
相关产品推荐
相关产品推荐

