如何在FastAPI中通过add_api_route传递请求模型构建通用路由类

我正尝试创建一个可处理任意数据库实体的基础路由类，实现新增、更新、删除等基础功能。由于类方法无法使用@router.get("/something")这类装饰器，所以用add_api_route函数替代，但遇到了如何将初始化时传入的模型传递给新增路由的问题。

我的代码如下：

from fastapi import APIRouter

from domain.services.base import BaseService

"""
    Base Router is very basic and just create a name and a router.
"""
class BaseRouter():
    def __init__(self, name):
        self.name = name
        self.router = APIRouter(tags=[self.name])

"""
    Basic Router inherit's BaseRouter, it's added functionality is that it handles getting, updating, and deleting
    with any type of entity.
"""
class BasicRouter(BaseRouter):
    def __init__(self, name, entity, model):    
        super().__init__(name)

        self.service = BaseService(name, entity, model)
        self.entity = entity
        self.model = model

        # register routes
        self.router.add_api_route(f"/{self.name}/Get", self.get, methods=["GET"])        
        self.router.add_api_route(f"/{self.name}/Add", self.add, methods=["POST"], response_model=self.model)
        self.router.add_api_route(f"/{self.name}/Update", self.update, methods=["PUT"])
        self.router.add_api_route(f"/{self.name}/Delete", self.delete, methods=["DELETE"])

    async def get(self, id):
        return self.service.getById(id)

    def add(self):
        pass

    def update(self):
        pass
    
    def delete(self):
        pass

我已尝试过这些方法：

• 查看add_api_route的源码寻找请求模型参数
• 搜索FastAPI官方文档中关于该函数的内容
• 在Flask和FastAPI的Discord服务器咨询

方法一：用闭包绑定请求模型生成处理函数 #

FastAPI需要在接口处理函数中明确声明请求体模型，你可以通过闭包动态绑定初始化时传入的model，生成带模型参数的处理函数，再注册到路由中。

修改后的BasicRouter类示例：

from fastapi import APIRouter, Body

from domain.services.base import BaseService

class BaseRouter():
    def __init__(self, name):
        self.name = name
        self.router = APIRouter(tags=[self.name])

class BasicRouter(BaseRouter):
    def __init__(self, name, entity, model):    
        super().__init__(name)

        self.service = BaseService(name, entity, model)
        self.entity = entity
        self.model = model

        # 注册路由时绑定动态生成的处理函数
        self.router.add_api_route(f"/{self.name}/Get", self.get, methods=["GET"])        
        self.router.add_api_route(
            f"/{self.name}/Add", 
            self._create_add_handler(), 
            methods=["POST"], 
            response_model=self.model
        )
        self.router.add_api_route(
            f"/{self.name}/Update", 
            self._create_update_handler(), 
            methods=["PUT"],
            response_model=self.model
        )
        self.router.add_api_route(f"/{self.name}/Delete", self.delete, methods=["DELETE"])

    async def get(self, id: int):
        return await self.service.getById(id)

    def _create_add_handler(self):
        # 闭包绑定当前实例的model
        async def add_handler(item: self.model):
            return await self.service.add(item.dict())
        return add_handler

    def _create_update_handler(self):
        async def update_handler(id: int, item: self.model = Body(...)):
            return await self.service.update(id, item.dict())
        return update_handler
    
    async def delete(self, id: int):
        return await self.service.delete(id)

方法二：利用路由状态+依赖注入传递模型 #

将模型和服务实例存入路由的状态中，通过依赖注入在处理函数中获取，适合需要更灵活扩展的场景：

from fastapi import APIRouter, Request, Body

from domain.services.base import BaseService

# 定义依赖获取模型
async def get_current_model(request: Request):
    return request.state.model

# 定义依赖获取服务
async def get_current_service(request: Request):
    return request.state.service

class BaseRouter():
    def __init__(self, name):
        self.name = name
        self.router = APIRouter(tags=[self.name])

class BasicRouter(BaseRouter):
    def __init__(self, name, entity, model):    
        super().__init__(name)

        self.service = BaseService(name, entity, model)
        self.entity = entity
        self.model = model

        # 给路由状态赋值
        self.router.state.model = model
        self.router.state.service = self.service

        # 注册路由
        self.router.add_api_route(f"/{self.name}/Get", self.get, methods=["GET"])        
        self.router.add_api_route(f"/{self.name}/Add", self.add, methods=["POST"], response_model=self.model)
        self.router.add_api_route(f"/{self.name}/Update", self.update, methods=["PUT"], response_model=self.model)
        self.router.add_api_route(f"/{self.name}/Delete", self.delete, methods=["DELETE"])

    async def get(self, id: int):
        return await self.service.getById(id)

    async def add(self, item: get_current_model(), service: get_current_service()):
        return await service.add(item.dict())

    async def update(self, id: int, item: get_current_model() = Body(...), service: get_current_service()):
        return await service.update(id, item.dict())
    
    async def delete(self, id: int):
        return await self.service.delete(id)

更优实践：参考第三方CRUD路由实现 #

如果不想重复造轮子，可以参考fastapi-crudrouter这类第三方库的实现思路，它通过泛型和动态路由生成，直接支持任意ORM模型的CRUD接口，比手动实现更简洁稳定。

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