FastAPI中可通过依赖函数接收Request参数并读取headers实现动态分支,需避免缓存、确保per-request调用,推荐用Annotated+Depends声明,注意反向代理和CORS对header的影响。
FastAPI 的依赖注入本身不支持“运行时根据请求头切换依赖实例”,但可以通过在依赖函数内部读取 Request 对象的 headers 来实现逻辑分发。关键点是:**依赖函数必须声明 Request 类型参数,且不能提前被缓存为单例**。
Depends 包裹一个带 Request 参数的函数,而不是直接传入类或无参函数lru_cache 或模块级变量缓存返回值,否则请求头变化不会触发重新计算比如根据 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,且

Annotated[...] 是推荐写法,兼容性更好。
实际调试时经常发现 request.headers.get("X-Client-Type") 是 None,原因通常是:
curl -H "X-Client-Type: mobile" 验证)underscores_in_headers on;)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) 验证路由行为。
真正要注意的是:所有依赖分支最终返回的对象,其生命周期必须与当前请求对齐。如果内部持有了长连接、上下文变量或异步状态,记得确认它们不会跨请求泄漏。