FastAPI中可通过依赖函数接收Request参数并读取headers实现动态分支,需避免缓存、确保per-request调用,推荐用Annotated+Depends声明,注意反向代理和CORS对header的影响。

依赖注入中如何读取请求头做动态分支
FastAPI 的依赖注入本身不支持“运行时根据请求头切换依赖实例”,但可以通过在依赖函数内部读取 Request 对象的 headers 来实现逻辑分发。关键点是:**依赖函数必须声明 Request 类型参数,且不能提前被缓存为单例**。
- 必须用
Depends包裹一个带Request参数的函数,而不是直接传入类或无参函数 - 避免使用
lru_cache或模块级变量缓存返回值,否则请求头变化不会触发重新计算 - FastAPI 默认对每个请求都重新调用依赖函数,所以只要不手动缓存,天然支持 per-request 分支
写一个能读 Header 并返回不同服务实例的依赖
比如根据 X-Client-Type: mobile 或 web 返回不同的数据库连接池或认证策略。示例中返回不同行为的 UserService 实例:
from fastapi import Depends, Request from typing import Annotatedclass UserService: def init(self, mode: str): self.mode = mode
def get_user_service(request: Request) -> UserService: client_type = request.headers.get("X-Client-Type", "web").lower() if client_type == "mobile": return UserService(mode="mobile-optimized") elif client_type == "desktop": return UserService(mode="desktop-heavy") else: return UserService(mode="default")
在路由中使用
@app.get("/users") def list_users(service: Annotated[UserService, Depends(get_user_service)]): return {"mode": service.mode}
注意:get_user_service 必须接收 Request,且不能加默认值(否则 FastAPI 无法注入);Annotated[...] 是推荐写法,兼容性更好。
常见踩坑:Header 读不到或总是走默认分支
实际调试时经常发现 request.headers.get("X-Client-Type") 是 None,原因通常是:
- 客户端根本没发这个 header(用
curl -H "X-Client-Type: mobile"验证) - 反向代理(如 Nginx)过滤或重命名了 header(Nginx 默认不透传带下划线的 header,需配
underscores_in_headers on;) -
浏览器 CORS 预检失败,导致带自定义 header 的请求根本没到 FastAPI(检查预检响应是否含
Access-Control-Allow-Headers) - 依赖函数被意外提升为全局单例——例如写成
service = get_user_service(...)赋值到模块顶层
需要复用逻辑?封装成可配置的工厂函数
如果多个依赖都要按 header 分支,别重复写 if/else,抽成工厂:
def make_header_router(
header_name: str,
mapping: dict[str, callable],
default_factory: callable
):
def router(request: Request):
value = request.headers.get(header_name, "").strip().lower()
factory = mapping.get(value, default_factory)
return factory(request)
return router
使用
mobile_db = lambda r: Database(url="mobile.db")
web_db = lambda r: Database(url="web.db")
get_db = make_header_router(
header_name="X-DB-Target",
mapping={"mobile": mobile_db, "web": web_db},
default_factory=web_db
)
这种写法把分支逻辑和具体实现解耦,也方便单元测试——你可以直接调用 get_db(fake_request) 验证路由行为。
真正要注意的是:所有依赖分支最终返回的对象,其生命周期必须与当前请求对齐。如果内部持有了长连接、上下文变量或异步状态,记得确认它们不会跨请求泄漏。










