17370845950

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

依赖注入中如何读取请求头做动态分支

FastAPI 的依赖注入本身不支持“运行时根据请求头切换依赖实例”,但可以通过在依赖函数内部读取 Request 对象的 headers 来实现逻辑分发。关键点是:**依赖函数必须声明 Request 类型参数,且不能提前被缓存为单例**。

  • 必须用 Depends 包裹一个带 Request 参数的函数,而不是直接传入类或无参函数
  • 避免使用 lru_cache 或模块级变量缓存返回值,否则请求头变化不会触发重新计算
  • FastAPI 默认对每个请求都重新调用依赖函数,所以只要不手动缓存,天然支持 per-request 分支

写一个能读 Header 并返回不同服务实例的依赖

比如根据 X-Client-Type: mobileweb 返回不同的数据库连接池或认证策略。示例中返回不同行为的 UserService 实例:

from fastapi import Depends, Request
from typing import Annotated

class 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) 验证路由行为。

真正要注意的是:所有依赖分支最终返回的对象,其生命周期必须与当前请求对齐。如果内部持有了长连接、上下文变量或异步状态,记得确认它们不会跨请求泄漏。