FastAPI Router用Union参数时无法解析Pydantic Enum类型求助

我定义了三个Pydantic模型，其中DocumentConfigBase包含枚举类型字段mode。在FastAPI路由中为knowledge_config参数使用Union[DocumentConfigCreate, SheetConfigCreate, ImageConfigCreate]类型后，调用API时出现错误："detail": "'config'"，且打印knowledge_config仅输出knowledge_id=1，其余参数丢失。

模型代码 #

import enum
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field, ConfigDict

class SplitType(str, enum.Enum):
    AUTO = "auto"
    CUSTOM = "custom"


class DocumentConfigBase(BaseModel):
    '''
    document config detail
    '''
    model_config = ConfigDict(use_enum_values=True)
    mode: Optional[SplitType] = Field(
        default=SplitType.AUTO, validate_default=True)
    chunk_size: Optional[str] = "800"
    identifier: Optional[str]
    text_preprocessing_rules: Dict[str, Any]


class DocumentConfigCreate(BaseModel):
    '''
    document config create
    '''
    knowledge_id: int
    config: DocumentConfigBase


class SheetConfigCreate(BaseModel):
    '''
    need to update
    '''
    knowledge_id: int
    pass


class ImageConfigCreate(BaseModel):
    '''
    need to update
    '''
    knowledge_id: int
    pass

路由代码 #

from fastapi import APIRouter, Body, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import Annotated, Union

router = APIRouter()

@router.post("/config", response_model=KnowledgeConfigModel)
def create_knowledge_config(
    user_id: Annotated[int, Body()], 
    config_type: Annotated[KnowledgeType, Body()], 
    knowledge_config: Union[DocumentConfigCreate, SheetConfigCreate, ImageConfigCreate], 
    session: Session = Depends(get_db)
):
    print('knowledge_config', knowledge_config)

    try:
        knowledge_model = KnowledgeConfigService.create(
            operator=user_id, config_type=config_type,
            knowledge_config=knowledge_config, session=session)
        print('knowledge_model', knowledge_model)
        if knowledge_model is None:
            raise HTTPException(
                status_code=404, detail="Config create failed")
        return knowledge_model
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e

请求参数 #

{
  "user_id": 0,
  "config_type": "document",
  "knowledge_config": {
    "knowledge_id": 1,
    "config": {
      "mode": "auto",
      "chunk_size": "1000",
      "identifier": "string",
      "text_preprocessing_rules": {}
    }
  }
}

• 不传递mode参数时，无错误，可获取所有传入参数，mode自动设为"auto"
• 移除Union类型，仅使用knowledge_config: DocumentConfigCreate时，无错误，可获取所有传入参数，mode自动设为"auto"

问题原因 #

FastAPI解析Union类型时会按顺序匹配模型，SheetConfigCreate和ImageConfigCreate仅包含knowledge_id字段，FastAPI发现请求数据中存在knowledge_id后，直接匹配了第一个符合的简单模型，忽略了后续的DocumentConfigCreate，导致config字段被丢弃，引发验证错误。

解决方案 #

方案1：使用Pydantic鉴别器（推荐）

给Union中的模型添加统一鉴别字段，让FastAPI明确匹配对应模型：

from typing import Literal
from pydantic import Discriminator, Annotated

class DocumentConfigCreate(BaseModel):
    knowledge_id: int
    config: DocumentConfigBase
    type: Literal["document"] = "document"

class SheetConfigCreate(BaseModel):
    knowledge_id: int
    type: Literal["sheet"] = "sheet"

class ImageConfigCreate(BaseModel):
    knowledge_id: int
    type: Literal["image"] = "image"

# 路由中修改参数定义
knowledge_config: Annotated[
    Union[DocumentConfigCreate, SheetConfigCreate, ImageConfigCreate],
    Discriminator("type")
]

请求时需带上type字段：

{
  "user_id": 0,
  "config_type": "document",
  "knowledge_config": {
    "type": "document",
    "knowledge_id": 1,
    "config": {
      "mode": "auto",
      "chunk_size": "1000",
      "identifier": "string",
      "text_preprocessing_rules": {}
    }
  }
}

方案2：调整Union模型顺序

将最复杂的DocumentConfigCreate放在Union首位，让FastAPI优先匹配：

knowledge_config: Union[DocumentConfigCreate, SheetConfigCreate, ImageConfigCreate]

注：该方法仅适用于模型字段无重叠冲突的场景，可靠性不如鉴别器。

方案3：根据config_type手动解析

利用已传入的config_type参数，在路由内手动选择模型解析请求体：

from fastapi import Request
import json

@router.post("/config", response_model=KnowledgeConfigModel)
async def create_knowledge_config(
    user_id: Annotated[int, Body()], 
    config_type: Annotated[KnowledgeType, Body()], 
    request: Request,
    session: Session = Depends(get_db)
):
    body = await request.json()
    knowledge_config_data = body.get("knowledge_config")
    
    if config_type == "document":
        knowledge_config = DocumentConfigCreate(**knowledge_config_data)
    elif config_type == "sheet":
        knowledge_config = SheetConfigCreate(**knowledge_config_data)
    elif config_type == "image":
        knowledge_config = ImageConfigCreate(**knowledge_config_data)
    else:
        raise HTTPException(status_code=400, detail="Invalid config_type")
        
    print('knowledge_config', knowledge_config)
    # 后续业务逻辑不变

内容的提问来源于stack exchange，提问作者Yu xin Dong