FastAPI启动阶段覆盖Pydantic模型后接口文档未更新的问题

在FastAPI应用中，需要创建一个作为输入参数的动态Pydantic模型，当启动阶段检查的某个条件为True时，覆盖该模型。目前的实现是在lifespan事件中重新赋值Model，但接口文档（/docs）仍然显示默认模型的参数限制，所有使用该模型的端点也没有同步更新。

最小复现代码如下：

from contextlib import asynccontextmanager
from typing import Annotated, Type

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, create_model

def dynamic_model(lim=10) -> Type[BaseModel]:
    return create_model("DynamicModel", lim=(Annotated[int, Query(ge=1, le=lim)], 1))

Model = dynamic_model()

@asynccontextmanager
async def lifespan(_app: FastAPI):
    global Model
    something_is_true = True
    if something_is_true:
        Model = dynamic_model(lim=100)  # 尝试覆盖模型
        
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def root(model: Annotated[Model, Depends()]):
    # 使用快捷方式通过查询参数调用端点
    return model

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", reload=True)

当前问题：示例代码未能成功覆盖模型，接口文档仍显示默认模型的参数限制。

方案一：启动前初始化模型（推荐） #

既然条件仅在启动阶段检查，直接在创建FastAPI实例前完成模型的初始化，这样FastAPI生成OpenAPI Schema时就会使用正确的模型，无需后续修改。

修改后的代码：

from typing import Annotated, Type

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, create_model

def dynamic_model(lim=10) -> Type[BaseModel]:
    return create_model("DynamicModel", lim=(Annotated[int, Query(ge=1, le=lim)], 1))

# 启动前检查条件并初始化模型
something_is_true = True
Model = dynamic_model(lim=100) if something_is_true else dynamic_model()

app = FastAPI()

@app.get("/")
async def root(model: Annotated[Model, Depends()]):
    return model

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", reload=True)

方案二：动态依赖+手动更新Schema（适用于条件必须在lifespan中判断的场景） #

如果条件必须在lifespan内检查（比如依赖启动时加载的资源或配置），可以通过依赖函数动态返回模型，并手动更新OpenAPI Schema：

修改后的代码：

from contextlib import asynccontextmanager
from typing import Annotated, Type

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, create_model

def dynamic_model(lim=10) -> Type[BaseModel]:
    return create_model("DynamicModel", lim=(Annotated[int, Query(ge=1, le=lim)], 1))

# 存储当前使用的模型
current_model = dynamic_model()

def get_dynamic_model() -> Type[BaseModel]:
    return current_model

@asynccontextmanager
async def lifespan(app: FastAPI):
    global current_model
    something_is_true = True
    if something_is_true:
        current_model = dynamic_model(lim=100)
        # 更新所有使用该模型的端点参数定义
        for route in app.routes:
            if hasattr(route, "dependant"):
                for param in route.dependant.params:
                    # 找到绑定了动态模型依赖的参数
                    if param.dependency == get_dynamic_model:
                        param.type_ = current_model
                        # 更新字段的校验规则
                        param.field_info = current_model.__fields__["lim"].field_info
        # 重新生成OpenAPI Schema
        app.openapi()
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root(model: Annotated[BaseModel, Depends(get_dynamic_model)]):
    return model

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", reload=True)

FastAPI在启动时会根据端点定义生成OpenAPI Schema，此时已经将初始的Model类型绑定到端点和Schema中。lifespan事件是在应用启动后才执行的，修改全局变量Model不会改变已经生成的Schema和端点绑定的类型，因此接口文档和参数校验都不会同步更新。

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