如何为一组Pydantic模型动态创建FastAPI路由/处理函数?

2 投票
1 回答
130 浏览
提问于 2025-04-14 18:05

我在使用FastAPI的时候遇到了一个问题,想要动态定义路由。结果发现最后定义的路由/处理函数会覆盖之前的所有定义(见下图)。

情况是这样的:我有很多Pydantic模型,都是从models.py导入的(在这个例子中简化了),我想为每个模型创建一个GET接口。

大概是这样的(我的应用更复杂——这样设计是有原因的——但为了找到问题,我把它简化成这样,问题依然存在):

# models.py

class Thing(BaseNode):
    value: str

class Other(BaseNode):
    value: str

model_list = [Thing, Other]



# app.py
from models import model_list

app = FastApi()


# Iterate over the models list and define a get function
for model in models_list:

    @app.get(f"/{model.__name__.lower()}")
    def get() -> str:
         print(model.__name__)
         return f"Getting {model.__name__)

调用GET Thing返回GET Other的响应

显然我做错了什么(或者FastAPI以某种奇怪的方式引用处理函数……是通过模块/函数名吗?所以get()函数在每次迭代时被重新定义了?)有没有什么想法,或者更好的结构方式?(我可不想为每个模型手动写一个新函数!)

谢谢!

更新:

如果我把get()函数定义在另一个函数里面,它就能正常工作:

for model in models_list:
     def get(model):
         def _get():
             print("Getting", model)
                
             return f"Getting {model.__name__}"
         return _get

     app.get(.get(f"/{model.__name__.lower()}", name=f"{model.__name__}.View")(get(model))

1 个回答

2

问题很可能出在你为每个创建的端点使用了相同的 model 引用,这样的话,它总是会被赋值为循环中最后一个值。

你应该使用一个辅助函数,它返回一个内部函数,就像在这个回答的第二个选项中展示的那样。

示例

from fastapi import FastAPI, APIRouter, Request
from pydantic import BaseModel


class One(BaseModel):
    value: str


class Two(BaseModel):
    value: str


app = FastAPI()
models = [One, Two]


def create_endpoint(m_name: str):
    async def endpoint(request: Request):
        print(request.url)
        return f"You called {m_name}"
    return endpoint


for m in models:    
    app.add_api_route(f"/{m.__name__.lower()}", create_endpoint(m.__name__), methods=["GET"])

撰写回答