FastAPI运行时动态更新枚举与OpenAPI规范方案问询

问题背景 #

我有一个向其他系统提供数据的FastAPI应用，数据按汽车厂商（如VW、Skoda等）分类。应用启动时会通过carHandler类扫描文件库，生成对应枚举模型用于接口参数声明，这个枚举会同步到openAPI.json，让用户能在SwaggerUI/OpenAPI UI的下拉菜单中选择厂商。

现在需要定期重新扫描文件库（会持续新增厂商），但目前必须重启FastAPI才能更新枚举。尝试过以下方法均无效：

• 直接在参数中使用函数调用：

  async def get_truck_manufacturer_information(name: cH.getCarModel(["trucks"]))
  

  报错：Call expression not allowed in type expressionPylancereportInvalidTypeForm
• 更新全局枚举变量：修改allowedTruckModel后，无法同步到FastAPI的OpenAPI schema和参数验证逻辑
• 尝试刷新FastAPI应用：编写接口重置openapi_schema并调用setup()，但无效果：

  @self.get("/updateCH", include_in_schema=False)
  async def updateCH(request: Request):
      cH.update()
      reloadKeys()
      updateModels()
      request.app.openapi_schema = None
      request.app.setup()
      return f"success"
  

核心代码片段：

class carHandler:
    ...
    def getCarModel(self, type) -> list:
        tempArr = []
        for manufacturer in self.manufacturers:
            if (manufacturer.type in type or type == ["all"]):
                tempArr.append((manufacturer.name, manufacturer.name))
        tempArrSorted = sorted(tempArr, key=lambda x: x[0])
        return enum.Enum(f'Available {type}', dict(tempArrSorted))

cH = carHandler()
allowedTruckModel = cH.getCarModel(["trucks"])
...
@self.get("/manufacturer/{name}", tags=["api"])
async def get_truck_manufacturer_information(name: allowedTruckModel) -> models.outputModel:
    return endpoint_logic.get_manufacturer(name)

问题根源 #

FastAPI在启动阶段会静态解析接口的类型注解，生成对应的OpenAPI schema和参数验证规则，之后不会自动重新解析。而Python的类型注解要求必须是静态类型（不能是运行时的函数调用结果），同时全局枚举变量更新后，FastAPI已经缓存了之前的验证逻辑和schema，不会主动更新。

解决方案 #

方案1：使用动态验证+动态生成OpenAPI枚举选项 #

放弃静态枚举，改用Annotated结合自定义验证函数，同时动态设置Query参数的enum属性来更新SwaggerUI的下拉选项。

1. 修改carHandler，维护当前厂商列表：

   class carHandler:
       def __init__(self):
           self.manufacturers = []
           self.update()  # 初始化扫描
   
       def update(self):
           # 扫描文件库更新self.manufacturers的逻辑
           ...
   
       def get_available_manufacturers(self, type_filter):
           # 返回符合类型的厂商名称列表
           return sorted(
               [m.name for m in self.manufacturers if m.type in type_filter or type_filter == ["all"]],
               key=str.lower
           )
   

2. 编写动态验证函数：

   def validate_manufacturer(name: str, type_filter):
       available = cH.get_available_manufacturers(type_filter)
       if name not in available:
           raise HTTPException(
               status_code=400,
               detail=f"Invalid manufacturer. Available options: {', '.join(available)}"
           )
       return name
   

3. 修改接口，使用动态枚举和验证：

   from fastapi import Query, HTTPException
   from typing import Annotated
   
   @self.get("/manufacturer/{name}", tags=["api"])
   async def get_truck_manufacturer_information(
       name: Annotated[str, Query(
           enum=cH.get_available_manufacturers(["trucks"]),
           description="Select a truck manufacturer"
       )]
   ) -> models.outputModel:
       validate_manufacturer(name, ["trucks"])
       return endpoint_logic.get_manufacturer(name)
   

4. 编写更新接口，刷新OpenAPI schema：

   @self.get("/updateCH", include_in_schema=False)
   async def updateCH(request: Request):
       cH.update()
       # 清除缓存的OpenAPI schema，下次访问会重新生成
       request.app.openapi_schema = None
       return "success"
   

   每次调用/updateCH后，下次访问SwaggerUI时，FastAPI会重新生成schema，此时Query的enum会使用最新的厂商列表；同时每次请求都会通过validate_manufacturer检查当前厂商是否有效，避免使用过期的枚举。

方案2：动态更新枚举并重新绑定端点（进阶） #

如果必须使用枚举类型，可以动态更新枚举的成员，同时重新生成OpenAPI schema。但需要注意，Python的枚举成员默认是不可变的，所以需要重新创建枚举实例并替换全局变量，同时让FastAPI重新识别验证规则。

1. 修改carHandler的getCarModel方法：

   def getCarModel(self, type_filter):
       tempArr = []
       for manufacturer in self.manufacturers:
           if manufacturer.type in type_filter or type_filter == ["all"]:
               tempArr.append((manufacturer.name, manufacturer.name))
       tempArrSorted = sorted(tempArr, key=lambda x: x[0])
       # 每次都创建新的枚举实例
       return enum.Enum(f'Available_{"_".join(type_filter)}', dict(tempArrSorted))
   

2. 全局变量存储枚举，编写更新函数：

   allowedTruckModel = cH.getCarModel(["trucks"])
   
   def update_truck_enum():
       global allowedTruckModel
       allowedTruckModel = cH.getCarModel(["trucks"])
   

3. 使用依赖项动态获取当前枚举：

   from fastapi import Depends
   
   def get_truck_manufacturer_enum():
       return allowedTruckModel
   
   @self.get("/manufacturer/{name}", tags=["api"])
   async def get_truck_manufacturer_information(
       name: Annotated[allowedTruckModel, Depends(get_truck_manufacturer_enum)]
   ) -> models.outputModel:
       return endpoint_logic.get_manufacturer(name)
   

4. 更新接口修改+自定义OpenAPI生成函数：

   @self.get("/updateCH", include_in_schema=False)
   async def updateCH(request: Request):
       cH.update()
       update_truck_enum()
       request.app.openapi_schema = None
       return "success"
   
   # 自定义OpenAPI生成函数，确保枚举是最新的
   def custom_openapi(app):
       if app.openapi_schema:
           return app.openapi_schema
       openapi_schema = get_openapi(
           title="Your API Title",
           version="1.0",
           routes=app.routes,
       )
       # 手动更新路径参数的枚举选项
       path_schema = openapi_schema["paths"]["/manufacturer/{name}"]["get"]
       for param in path_schema["parameters"]:
           if param["name"] == "name":
               param["schema"]["enum"] = [member.value for member in allowedTruckModel]
       app.openapi_schema = openapi_schema
       return openapi_schema
   
   app.openapi = lambda: custom_openapi(app)
   

总结 #

方案1更简洁可靠，无需依赖静态枚举，通过动态验证和动态生成Query枚举选项实现需求；方案2适合必须使用枚举类型的场景，但需要更复杂的schema自定义逻辑。

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