在FastAPI项目中,依赖注入是核心特性之一,很多时候我们需要根据请求头中的不同信息,动态切换依赖的具体实现,比如根据请求头里的客户端版本号加载不同版本的业务逻辑,或者根据认证头选择不同的权限校验方式。

基础实现:通过header参数判断切换
最直接的方式是在依赖函数中接收请求头参数,根据参数值返回不同的实例或执行不同的逻辑。首先我们定义一个简单的业务接口和两个不同版本的实现类:
from fastapi import FastAPI, Depends, Header
from typing import Optional
app = FastAPI()
# 定义两个不同版本的服务类
class V1Service:
def get_data(self):
return {"version": "v1", "data": "v1版本的数据"}
class V2Service:
def get_data(self):
return {"version": "v2", "data": "v2版本的数据"}
# 依赖函数,根据header中的version参数切换服务实例
def get_dynamic_service(version: Optional[str] = Header(None, alias="X-Version")):
if version == "v2":
return V2Service()
# 默认返回v1版本的服务
return V1Service()
@app.get("/data")
def read_data(service: V1Service | V2Service = Depends(get_dynamic_service)):
return service.get_data()
上面的代码中,我们定义了一个get_dynamic_service依赖,它接收请求头中的X-Version参数,根据参数值返回不同的服务实例。当请求头携带X-Version: v2时,返回v2版本的服务,否则返回默认的v1版本服务。
进阶实现:使用依赖工厂模式
如果动态切换的逻辑比较复杂,或者需要支持更多的切换条件,可以使用依赖工厂模式,将切换逻辑封装起来,提高代码的可维护性。示例如下:
from fastapi import FastAPI, Depends, Header
from typing import Callable, Optional
app = FastAPI()
class BaseService:
def process(self):
raise NotImplementedError
class UserService(BaseService):
def process(self):
return {"service_type": "user", "result": "用户服务处理结果"}
class OrderService(BaseService):
def process(self):
return {"service_type": "order", "result": "订单服务处理结果"}
# 依赖工厂函数,根据header返回对应的依赖
def create_service_dependency(service_type: str):
def get_service(service_type_header: Optional[str] = Header(None, alias="X-Service-Type")):
# 优先使用依赖工厂传入的默认类型,再使用请求头中的类型
final_type = service_type_header if service_type_header else service_type
if final_type == "order":
return OrderService()
return UserService()
return get_service
# 使用工厂创建依赖,默认使用user服务
user_service_dependency = create_service_dependency("user")
@app.get("/process")
def process_data(service: BaseService = Depends(user_service_dependency)):
return service.process()
这种方式下,我们可以通过工厂函数预先定义依赖的默认行为,同时支持请求头覆盖默认配置,适合需要在多个接口复用相同切换逻辑的场景。
嵌套依赖实现复杂切换逻辑
当切换逻辑需要结合多个header参数,或者需要先经过其他依赖处理时,可以使用嵌套依赖的方式。比如我们需要先校验请求头中的token,再根据角色切换服务:
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional
app = FastAPI()
# 校验token的依赖
def verify_token(token: Optional[str] = Header(None, alias="X-Token")):
if not token:
raise HTTPException(status_code=401, detail="缺少token")
if token != "valid_token":
raise HTTPException(status_code=401, detail="token无效")
return token
# 根据角色切换服务的依赖,嵌套了token校验依赖
def get_service_by_role(
token: str = Depends(verify_token),
role: Optional[str] = Header(None, alias="X-Role")
):
if role == "admin":
return {"role": "admin", "permission": "all"}
return {"role": "normal", "permission": "read"}
@app.get("/info")
def get_info(info: dict = Depends(get_service_by_role)):
return info
这里的get_service_by_role依赖首先依赖verify_token完成token校验,再根据X-Role请求头返回不同的信息,实现了多步骤的动态切换逻辑。
注意事项
- 请求头参数的别名需要和实际请求头名称一致,FastAPI默认会将下划线转换为连字符,所以如果使用
Header(..., alias="X-Version")才能正确匹配X-Version请求头。 - 动态切换的依赖如果返回不同类型的实例,需要注意接口参数类型注解的兼容性,避免类型检查报错。
- 如果切换逻辑涉及复杂的业务判断,建议将判断逻辑抽成独立函数,保持依赖函数的简洁性。
动态切换依赖的核心思路是在依赖函数内部获取请求头参数,根据参数值返回不同的结果,结合工厂模式和嵌套依赖可以应对更复杂的业务场景。