如何判断FastAPI中Pydantic联合响应模型的实际返回类型

要在FastAPI路由内部判断返回的是哪个Pydantic模型并完成解析，核心思路是利用模型的特征差异做区分，下面是几种实用的实现方式：

1. 通过专属特征字段判断（推荐） #

给每个模型定义一个唯一的标识字段（比如status），通过这个字段直接匹配对应的模型：

首先定义带特征字段的Pydantic模型：

from pydantic import BaseModel
from typing import Union

class SuccessResponse(BaseModel):
    status: str = "success"
    data: dict

class FailureResponse(BaseModel):
    status: str = "failure"
    message: str

class ErrorResponse(BaseModel):
    status: str = "error"
    error_code: int
    detail: str

然后在路由中根据特征字段选择解析模型：

from fastapi import FastAPI

app = FastAPI()

@app.get("/demo", response_model=Union[SuccessResponse, FailureResponse, ErrorResponse])
async def demo_endpoint():
    # 模拟从外部服务/数据库获取的原始响应数据
    raw_response = {"status": "success", "data": {"user_id": 123, "username": "abinav"}}
    
    # 根据status字段匹配对应模型
    match raw_response.get("status"):
        case "success":
            parsed_data = SuccessResponse.parse_obj(raw_response)
        case "failure":
            parsed_data = FailureResponse.parse_obj(raw_response)
        case "error":
            parsed_data = ErrorResponse.parse_obj(raw_response)
        case _:
            raise ValueError("无法识别的响应类型")
    
    # 后续可直接使用parsed_data的结构化属性
    print(parsed_data.data["user_id"])
    return parsed_data

2. 封装通用解析函数（复用场景） #

如果多个路由都需要处理这类解析，可以把判断逻辑封装成通用函数，避免重复代码：

def parse_response(raw_data: dict):
    # 建立特征字段到模型的映射
    model_mapping = {
        "success": SuccessResponse,
        "failure": FailureResponse,
        "error": ErrorResponse
    }
    
    status = raw_data.get("status")
    target_model = model_mapping.get(status)
    if not target_model:
        raise ValueError(f"不支持的响应状态: {status}")
    
    return target_model.parse_obj(raw_data)

# 路由中调用
@app.get("/reusable-demo", response_model=Union[SuccessResponse, FailureResponse, ErrorResponse])
async def reusable_demo():
    raw_response = {"status": "error", "error_code": 400, "detail": "参数缺失"}
    parsed_data = parse_response(raw_response)
    return parsed_data

3. 异常捕获式解析（不推荐） #

如果模型之间字段差异足够大，也可以通过try-except依次尝试解析，但这种方式可读性差、出错概率高，仅适合字段完全无重叠的场景：

@app.get("/fallback-demo", response_model=Union[SuccessResponse, FailureResponse, ErrorResponse])
async def fallback_demo():
    raw_response = {"status": "failure", "message": "操作失败"}
    
    try:
        parsed_data = SuccessResponse.parse_obj(raw_response)
    except:
        try:
            parsed_data = FailureResponse.parse_obj(raw_response)
        except:
            parsed_data = ErrorResponse.parse_obj(raw_response)
    
    return parsed_data

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