在 Pydantic v2 中使用 bson.ObjectId

I found 一些例子关于如何在其中使用 ObjectIdBaseModel类。基本上，这可以通过创建 Pydantic 友好的类来实现，如下所示：

class PyObjectId(ObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if not ObjectId.is_valid(v):
            raise ValueError("Invalid objectid")
        return ObjectId(v)

    @classmethod
    def __modify_schema__(cls, field_schema):
        field_schema.update(type="string")

然而，这似乎适用于 Pydantic v1，因为该机制已被__get_pydantic_core_schema__类方法。但是，我无法使用 Pydantic v2 实现等效的解决方案。有可能吗？我需要什么验证器？我尝试重构一些东西，但无法得到任何可用的东西。

迁移旧式PyObjectId到最新的pydantic-v2版本，最简单的方法是使用带注释的验证器.

from typing import Any
from typing import Annotated, Union
from bson import ObjectId
from pydantic import PlainSerializer, AfterValidator, WithJsonSchema

def validate_object_id(v: Any) -> ObjectId:
    if isinstance(v, ObjectId):
        return v
    if ObjectId.is_valid(v):
        return ObjectId(v)
    raise ValueError("Invalid ObjectId")

PyObjectId = Annotated[
    Union[str, ObjectId],
    AfterValidator(validate_object_id),
    PlainSerializer(lambda x: str(x), return_type=str),
    WithJsonSchema({"type": "string"}, mode="serialization"),
]

然后您可以通过以下方式在模型中使用它：

from pydantic import BaseModel
from pydantic import ConfigDict, Field

class MyCustomModel(BaseModel):
    id: PyObjectId = Field(alias="_id")

    model_config = ConfigDict(arbitrary_types_allowed=True)

使用测试一下类型适配器:

import pytest
from bson import ObjectId
from pydantic import TypeAdapter, ConfigDict
    
@pytest.mark.parametrize("obj", ["64b7992ba8f08069073f1055", ObjectId("64b7992ba8f08069073f1055")])
def test_pyobjectid_validation(obj):
    ta = TypeAdapter(PyObjectId, config=ConfigDict(arbitrary_types_allowed=True))
    ta.validate_python(obj)

@pytest.mark.parametrize("obj", ["64b7992ba8f08069073f1055", ObjectId("64b7992ba8f08069073f1055")])
def test_pyobjectid_serialization(obj):
    ta = TypeAdapter(PyObjectId, config=ConfigDict(arbitrary_types_allowed=True))
    ta.dump_json(obj)

即使使用最新的解决方案，该解决方案也能正常工作FastAPIv0.100.0+