FastAPI迁移Pydantic V2后自定义ObjectId查询参数报错解决

问题分析 #

你遇到的报错虽提示与响应字段相关，但实际原因是FastAPI处理查询参数时，无法将bson.ObjectId识别为有效的Pydantic字段类型——即便用Annotated添加了校验器和序列化器，FastAPI构建请求模型时仍未正确适配该类型。

解决方案 #

在Pydantic V2中，推荐通过自定义类型+实现__get_pydantic_core_schema__方法扩展类型，这种方式能被FastAPI正确识别并处理参数的校验、序列化逻辑。

完整可行代码 #

import bson
from pydantic import GetCoreSchemaHandler
from pydantic_core import core_schema
from fastapi import FastAPI

class ObjectId(bson.ObjectId):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        # 定义核心Schema：允许输入为str或bson.ObjectId类型
        return core_schema.no_info_wrap_validator_function(
            cls.validate,
            core_schema.union_schema([
                core_schema.is_instance_schema(bson.ObjectId),
                core_schema.str_schema()
            ]),
            # 指定序列化逻辑：将ObjectId转为字符串
            serialization=core_schema.plain_serializer_function_ser_schema(
                lambda x: str(x), return_type=str
            )
        )
    
    @classmethod
    def validate(cls, value, handler):
        if isinstance(value, str):
            try:
                return bson.ObjectId(value)
            except bson.errors.InvalidId:
                raise ValueError("Invalid ObjectId")
        elif isinstance(value, bson.ObjectId):
            return value
        else:
            raise TypeError("ObjectId required")

app = FastAPI()

@app.get("/test")
def test(id: ObjectId) -> bool:
    return True

方案说明 #

1. 自定义ObjectId类：继承自bson.ObjectId，保证类型兼容性。
2. 核心Schema定义：通过__get_pydantic_core_schema__告诉Pydantic如何处理该类型：
   • 允许输入为字符串或原生bson.ObjectId
   • 绑定校验函数，自动将合法字符串转为ObjectId实例
   • 指定序列化规则，将ObjectId转为字符串（适配API的JSON格式）
3. FastAPI自动适配：自定义类符合Pydantic V2的类型扩展规范，FastAPI能正确解析查询参数，不会触发类型无效报错。

简化替代方案（全局配置） #

若不想自定义类，也可全局开启Pydantic的任意类型支持，结合Annotated使用：

import bson
from typing import Annotated
from pydantic import BeforeValidator, PlainSerializer, WithJsonSchema
from fastapi import FastAPI

ObjectId = Annotated[
    bson.ObjectId,
    BeforeValidator(lambda x: bson.ObjectId(x) if isinstance(x, str) else x),
    PlainSerializer(lambda x: f"{x}", return_type=str),
    WithJsonSchema({"type": "string"}, mode="validation"),
    WithJsonSchema({"type": "string"}, mode="serialization"),
]

# 全局开启任意类型允许配置
app = FastAPI(pydantic_config={"arbitrary_types_allowed": True})

@app.get("/test")
def test(id: ObjectId) -> bool:
    return True

但该方式全局配置可能带来潜在类型风险，优先推荐自定义类型方案。

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