如何为FastAPI端点动态创建基于依赖的验证模型以解耦控制器与依赖？

要解除MediaController与MediaPipeline的耦合，核心是动态生成请求模型，让控制器实例根据传入的管道列表自动适配对应的参数类型。以下是具体实现方案：

步骤1：统一管道参数类的命名 #

先调整所有MediaPipeline的内部参数类，统一命名为Params，方便后续批量获取：

from fastapi import FastAPI, APIRouter
from pydantic import BaseModel, Field
from typing import Literal, Union, Any

class MediaPipeline1():
    class Params(BaseModel):
        pipeline_name: Literal["One"]
        foo: int = 11

class MediaPipeline2():
    class Params(BaseModel):
        pipeline_name: Literal["Two"]
        bar: int = 22

class MediaPipeline3():
    class Params(BaseModel):
        pipeline_name: Literal["Three"]
        baz: int = 33

步骤2：动态生成请求模型的控制器 #

修改MediaController，在初始化时根据传入的管道列表，动态创建对应的RequestModel：

class MediaController():
    def __init__(self, pipelines: list):
        self.pipelines = pipelines
        # 收集所有管道的参数类
        param_classes = [pipe.Params for pipe in pipelines]
        # 构建动态Union类型
        params_union = Union[tuple(param_classes)]
        # 动态生成RequestModel类
        self.RequestModel = type(
            "DynamicRequestModel",
            (BaseModel,),
            {
                "params": Field(..., discriminator="pipeline_name"),
                "__annotations__": {"params": params_union}
            }
        )

    def handler(self, request: Any):
        # 实际处理逻辑：根据pipeline_name分发到对应管道
        for pipe in self.pipelines:
            if request.params.pipeline_name == pipe.Params.__fields__["pipeline_name"].default:
                print(f"Handling with {pipe.__class__.__name__}: {request.params}")
                break

步骤3：添加路由时指定动态请求模型 #

利用FastAPI的add_api_route支持request_model参数的特性，在注册路由时传入控制器实例的动态请求模型：

my_pipeline_1 = MediaPipeline1()
my_pipeline_2 = MediaPipeline2()
my_pipeline_3 = MediaPipeline3()

router = APIRouter()

# 控制器1：处理管道1和2
my_controller_12 = MediaController([my_pipeline_1, my_pipeline_2])
router.add_api_route("/1", my_controller_12.handler, request_model=my_controller_12.RequestModel)

# 控制器2：处理管道2和3
my_controller_23 = MediaController([my_pipeline_2, my_pipeline_3])
router.add_api_route("/2", my_controller_23.handler, request_model=my_controller_23.RequestModel)

app = FastAPI()
app.include_router(router)

关键说明 #

• 动态类型生成：通过type()函数在运行时创建RequestModel，彻底避免硬编码Union类型的耦合问题。
• FastAPI路由适配：add_api_route的request_model参数可直接指定动态生成的模型，FastAPI会自动完成请求的解析与验证。
• 解耦效果：现在MediaController可以接收任意数量的MediaPipeline实例，无需修改控制器内部代码，完全符合依赖注入的设计原则。

内容的提问来源于stack exchange，提问作者Multisync