FastAPI类式通用CRUD路由：如何实现动态请求模型与Swagger UI生成

我正在FastAPI中编写基于类的通用CRUD路由/处理器，需要一种类似response_model的请求模型机制——无需为每个端点编写单独函数，就能为输入模型生成Swagger UI。以下是我设想的读取用户模型并基于用户名过滤的示例代码，但这段代码无法运行：核心问题是无法在ModelReadHandler.read()方法中把self.request_model_cls作为参数类型提示。

我的目标是动态生成端点，且Swagger UI能正确显示其请求参数，同时不想采用为每个模型扩展类并添加类型提示的替代方案。

示例代码：

class ModelReadHandler:
    def __init__(self, base_url: str, response_model_cls: type, request_model_cls: type):
        self.router = APIRouter()
        self.response_model = response_model_cls
        self.request_model = request_model_cls
        self.add_routes()
    
    def add_routes(self):
        self.router.add_api_route(f'/{self.base_url}/', self.read, methods=['GET'],
                                   response_model=self.response_model)
    
    def read(self, input_model: self.request_model_cls  # 此处无法用实例属性作为类型提示
             session: Depends(get_session),
             ):
        query = input_model.get_read_query()
        return session.exec(query).all()

# 输入模型的抽象基类
class InputModel(pydantic.BaseModel):
    def get_read_query(self):
        raise NotImplementedError()

# 示例输入模型
class UserInputModel(InputModel):
    username: str
    def get_read_query(self):
        return select(UserModel).where(UserModel.username == self.username)

# 实例化处理器
user_handler = ModelReadHandler(base_url='user',
                                request_model_cls=UserInputModel,
                                response_model_cls=UserModel)

app = FastAPI()
app.include_router(user_handler.router)

FastAPI依赖函数的类型注解生成文档和解析请求，只要动态为视图函数设置正确的类型注解，就能实现需求。以下是两种可行方案：

方法1：动态依赖注入解析请求模型 #

通过创建动态依赖函数，为其添加请求模型的字段注解，让FastAPI自动解析参数并生成Swagger文档：

from fastapi import FastAPI, APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session
import pydantic
from typing import Type

# 示例数据库模型
class UserModel(pydantic.BaseModel):
    username: str

# 模拟数据库会话获取
def get_db():
    session = Session()
    try:
        yield session
    finally:
        session.close()

class InputModel(pydantic.BaseModel):
    @classmethod
    def from_query_params(cls, **kwargs):
        return cls(**kwargs)
    
    def get_read_query(self):
        raise NotImplementedError()

class UserInputModel(InputModel):
    username: str
    def get_read_query(self):
        return select(UserModel).where(UserModel.username == self.username)

class ModelReadHandler:
    def __init__(self, base_url: str, response_model_cls: Type[pydantic.BaseModel], request_model_cls: Type[InputModel]):
        self.router = APIRouter()
        self.response_model = response_model_cls
        self.request_model = request_model_cls
        self.add_routes()
    
    def add_routes(self):
        # 动态创建依赖函数，用于解析请求参数
        def request_model_dependency(**kwargs):
            return self.request_model.from_query_params(**kwargs)
        
        # 为依赖函数添加字段注解，让FastAPI识别参数
        request_model_dependency.__annotations__ = {
            field: field_info.annotation 
            for field, field_info in self.request_model.model_fields.items()
        }
        
        # 定义视图函数并绑定依赖
        def read(input_model: self.request_model = Depends(request_model_dependency),
                 session: Session = Depends(get_db)):
            query = input_model.get_read_query()
            return session.exec(query).all()
        
        self.router.add_api_route(f'/{self.base_url}/', read, methods=['GET'], response_model=list[self.response_model])

# 实例化处理器
user_handler = ModelReadHandler(base_url='user',
                                request_model_cls=UserInputModel,
                                response_model_cls=UserModel)

app = FastAPI()
app.include_router(user_handler.router)

方法2：动态修改函数签名与注解 #

直接为生成的视图函数设置正确的类型注解和签名，让FastAPI识别请求模型：

from fastapi import FastAPI, APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session
import pydantic
from typing import Type, List
import inspect

# 示例数据库模型
class UserModel(pydantic.BaseModel):
    username: str

# 模拟数据库会话获取
def get_db():
    session = Session()
    try:
        yield session
    finally:
        session.close()

class InputModel(pydantic.BaseModel):
    def get_read_query(self):
        raise NotImplementedError()

class UserInputModel(InputModel):
    username: str
    def get_read_query(self):
        return select(UserModel).where(UserModel.username == self.username)

class ModelReadHandler:
    def __init__(self, base_url: str, response_model_cls: Type[pydantic.BaseModel], request_model_cls: Type[InputModel]):
        self.router = APIRouter()
        self.response_model = response_model_cls
        self.request_model = request_model_cls
        self.add_routes()
    
    def add_routes(self):
        # 基础业务逻辑
        def read_logic(input_model: InputModel, session: Session):
            query = input_model.get_read_query()
            return session.exec(query).all()
        
        # 动态生成视图函数并修改注解
        read_func = lambda input_model, session: read_logic(input_model, session)
        read_func.__annotations__ = {
            'input_model': self.request_model,
            'session': Session,
            'return': List[self.response_model]
        }
        
        # 修改函数签名，让FastAPI正确识别参数和依赖
        read_func.__signature__ = inspect.signature(read_logic).replace(
            parameters=[
                inspect.Parameter('input_model', inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=self.request_model),
                inspect.Parameter('session', inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=Session, default=Depends(get_db))
            ]
        )
        
        self.router.add_api_route(
            f'/{self.base_url}/', 
            read_func, 
            methods=['GET'],
            response_model=List[self.response_model]
        )

# 实例化处理器
user_handler = ModelReadHandler(base_url='user',
                                request_model_cls=UserInputModel,
                                response_model_cls=UserModel)

app = FastAPI()
app.include_router(user_handler.router)

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