如何在FastAPI文档中动态生成非Enum类型的下拉选项？

我之前按照FastAPI官方文档的方法，用Enum类型实现了自动生成文档页面中的下拉选项，代码如下：

main.py 代码 #

from enum import Enum
from db import engine,SessionLocal
from models import Example,ExampleModel,Base,Color
from fastapi import FastAPI,Depends
from sqlalchemy.orm import Session
from sqlalchemy import select

app = FastAPI()

@app.on_event("startup")
def on_startup():
    Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/colors",response_model=list[ExampleModel])
def read_colors(color: Color ,db: Session = Depends(get_db)):
    return db.execute(select(Example).where(Example.colors == color)).scalars().all()

models.py 代码 #

from sqlalchemy.orm import Mapped,mapped_column,DeclarativeBase
from sqlalchemy import Identity,Enum
from pydantic import BaseModel, ConfigDict
import enum

class Base(DeclarativeBase):
    pass

class Color(str,enum.Enum):
    RED = 'RED'
    BLUE = 'BLUE'

class Example(Base):
    __tablename__='example'
    id: Mapped[int] = mapped_column(Identity(always=True),primary_key=True)
    colors: Mapped[Color] = mapped_column(Enum(Color))

class ExampleModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    colors: Color

这样实现后，Color类型能在文档页面正常渲染为下拉选项。但现在数据库表结构改成了下面这样，colors变成了文本字段，里面有10-15个不同值，而且这些值会随时间动态变化：

class Example(Base):
    __tablename__='example'
    id: Mapped[int] = mapped_column(Identity(always=True),primary_key=True)
    colors: Mapped[str] 

我想知道这种场景下能不能动态生成colors的下拉选项？

完全可以实现动态生成下拉选项，以下是两种实用方案：

方案一：动态生成Pydantic Enum（适合值更新不频繁的场景） #

在应用启动时从数据库拉取所有colors的唯一值，动态创建Enum类型作为接口参数类型。如果值会更新，可搭配定时任务刷新枚举。

核心代码示例 #

from enum import Enum
from sqlalchemy import distinct
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from db import engine, SessionLocal
from models import Example, ExampleModel, Base

app = FastAPI()

# 动态创建Color枚举的函数
def create_dynamic_color_enum(db: Session):
    color_values = db.execute(select(distinct(Example.colors))).scalars().all()
    return Enum('DynamicColor', {value.upper(): value for value in color_values}, type=str)

# 全局存储动态枚举，启动时初始化
DynamicColor = None

@app.on_event("startup")
def on_startup():
    Base.metadata.create_all(bind=engine)
    db = SessionLocal()
    global DynamicColor
    DynamicColor = create_dynamic_color_enum(db)
    db.close()

# 可选：添加定时任务刷新枚举（需安装APScheduler）
# from apscheduler.schedulers.asyncio import AsyncIOScheduler
# scheduler = AsyncIOScheduler()
# scheduler.add_job(refresh_color_enum, 'interval', minutes=30)
# scheduler.start()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/colors", response_model=list[ExampleModel])
def read_colors(color: DynamicColor, db: Session = Depends(get_db)):
    return db.execute(select(Example).where(Example.colors == color)).scalars().all()

方案二：修改OpenAPI规范（适合值频繁更新的场景） #

重写FastAPI的openapi()方法，每次生成文档时从数据库拉取最新的colors值，直接修改接口参数的schema，实现下拉选项的实时同步。

核心代码示例 #

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import select, distinct
from db import engine, SessionLocal
from models import Example, ExampleModel, Base

app = FastAPI()

@app.on_event("startup")
def on_startup():
    Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 自定义OpenAPI生成逻辑
def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = app.openapi()
    # 从数据库获取最新的color唯一值
    db = SessionLocal()
    color_values = db.execute(select(distinct(Example.colors))).scalars().all()
    db.close()
    # 给/colors接口的color参数添加枚举属性
    path = "/colors"
    if path in openapi_schema["paths"]:
        get_operation = openapi_schema["paths"][path]["get"]
        for param in get_operation["parameters"]:
            if param["name"] == "color":
                param["schema"]["enum"] = color_values
                param["schema"]["type"] = "string"
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

@app.get("/colors", response_model=list[ExampleModel])
def read_colors(color: str, db: Session = Depends(get_db)):
    return db.execute(select(Example).where(Example.colors == color)).scalars().all()

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