FastAPI可通过依赖注入结合Header值动态选择依赖实现:定义顶层依赖函数读取Header(如X-Mode),根据其值返回不同子依赖结果;也可通过Request对象获取多Header或复杂路由决策。

FastAPI 中可以通过依赖注入机制,结合请求头(Header)的值动态选择或切换依赖实现。核心思路是:定义一个顶层依赖函数,内部读取 Header 值(如 X-Auth-Type、User-Agent 或自定义标识),再根据该值返回不同实例或调用不同逻辑的子依赖。
使用依赖工厂函数动态返回具体依赖
这是最常用也最清晰的方式:把“选择逻辑”封装在依赖函数中,让它返回另一个依赖的执行结果。
- 定义多个底层依赖(例如
get_db_session、get_cache_client) - 编写一个顶层依赖(如
get_service_by_header),接收Request或显式声明的Header参数 - 在该函数中判断 Header 值,决定调用哪个具体服务
示例:
from fastapi import Depends, Header, Request
from typing import Annotated
<p>async def get_legacy_service():
return "LegacyService"</p><p>async def get_modern_service():
return "ModernService"</p><p>async def get_service_by_header(
x_mode: Annotated[str | None, Header(alias="X-Mode")] = None,
) -> str:
if x_mode == "modern":
return await get_modern_service()
else:
return await get_legacy_service()</p><p>@app.get("/data")
async def read_data(service: str = Depends(get_service_by_header)):
return {"service_used": service}
</p>通过 Request 对象读取 Header 并做更复杂路由
当需要访问多个 Header、校验格式、或结合路径/查询参数共同决策时,直接依赖 Request 更灵活。
-
Request可以获取所有原始请求信息,包括 headers、method、url 等 - Header 参数和
Request,FastAPI 会自动注入,无需额外处理 - ait>
示例:
from fastapi import Request, Depends
<p>async def get_service_by_request(request: Request) -> str:
mode = request.headers.get("X-Mode", "").lower()
user_agent = request.headers.get("User-Agent", "")</p><pre class="brush:php;toolbar:false;">if "mobile" in user_agent.lower():
return "MobileOptimizedService"
elif mode == "v2":
return "ApiV2Service"
else:
return "DefaultService"@app.get("/info") async def get_info(service: str = Depends(get_service_by_request)): return {"active_service": service}
配合 Class Dependency 实现状态化动态行为
如果依赖逻辑较重(比如初始化客户端、持有连接池),推荐用类依赖 + __call__ 方法,在每次请求时动态决定行为。
- __init__ 不应含业务逻辑(因为 FastAPI 可能复用实例)
- __call__ 中,它会在每次请求时被调用
- Request 或 Header,并返回不同实例
示例:
from fastapi import Request, Depends
<p>class DynamicClient:
def <strong>init</strong>(self):</p><h1>这里只做轻量初始化,不涉及请求上下文</h1><pre class="brush:php;toolbar:false;"> pass
async def __call__(self, request: Request) -> str:
auth_type = request.headers.get("X-Auth-Type", "basic")
if auth_type == "jwt":
return "JwtAuthenticatedClient"
elif auth_type == "api-key":
return "ApiKeyClient"
else:
return "BasicAuthClient"@app.get("/client") async def use_client(client: str = Depends(DynamicClient())): return {"client_type": client}
注意事项与避坑点
实际使用中容易忽略几个关键细节:
- Header 名称默认转为小写并用短横线连接(如
X-Api-Version→x_api_version),建议用alias显式声明 - 异步依赖必须用
await调用,或确保返回的是协程对象;同步函数可直接返回值 - 不要在依赖函数外层做 Header 判断(比如全局变量缓存),会导致并发下状态错乱
- 如果依赖返回的是可调用对象(如类实例),确保它本身支持被 FastAPI 正确调用(如实现
__call__或是标准依赖函数)
不复杂但容易忽略。










