使用 PyDantic v2.0^ 声明 Mongo ObjectId 的新方法是什么？

Question

本周，我开始使用 MongoDB 和 Flask，因此我找到了一篇有用的文章，介绍如何通过使用 PyDantic 库定义 MongoDB 的模型来将它们一起使用。然而，这篇文章有点过时了，大部分可以更新到新的 PyDantic 版本，但问题是 ObjectId 是第三方字段，并且在版本之间发生了很大的变化。

本文使用以下代码定义了ObjectId：

from bson import ObjectId
from pydantic.json import ENCODERS_BY_TYPE


class PydanticObjectId(ObjectId):
    """
    Object Id field. Compatible with Pydantic.
    """

    @classmethod
    def __get_validators__(cls):
        yield cls.validate
   
    #The validator is doing nothing
    @classmethod
    def validate(cls, v):
        return PydanticObjectId(v)

    #Here you modify the schema to tell it that it will work as an string
    @classmethod
    def __modify_schema__(cls, field_schema: dict):
        field_schema.update(
            type="string",
            examples=["5eb7cf5a86d9755df3a6c593", "5eb7cfb05e32e07750a1756a"],
        )

#Here you encode the ObjectId as a string
ENCODERS_BY_TYPE[PydanticObjectId] = str

在过去，这段代码运行良好。然而，我最近发现最新版本的 PyDantic 有一种更复杂的定义自定义数据类型的方法。我已经尝试遵循Pydantic 文档，但我仍然很困惑并且无法成功实现它。

我已经尝试过对第三方类型进行实现，但它不起作用。它与文档的代码几乎相同，但更改了字符串的整数以及 ObjectId 的第三方调用标签。再次，我不确定为什么它不起作用。

from bson import ObjectId
from pydantic_core import core_schema 
from typing import Annotated, Any
from pydantic import BaseModel, GetJsonSchemaHandler, ValidationError

from pydantic.json_schema import JsonSchemaValue


class PydanticObjectId(ObjectId):
    """
    Object Id field. Compatible with Pydantic.
    """

    x: str

    def __init__(self):
        self.x = ''

class _ObjectIdPydanticAnnotation:
    @classmethod
    def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: ObjectId[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:

        @classmethod
        def validate_object_id(cls, v: ObjectId) -> PydanticObjectId:
            if not ObjectId.is_valid(v):
                raise ValueError("Invalid objectid")
            return PydanticObjectId(v)
        
        from_str_schema = core_schema.chain_schema(
            [
                core_schema.str_schema(),
                core_schema.no_info_plain_validator_function(validate_object_id),
            ]
        )
        return core_schema.json_or_python_schema(
            json_schema=from_str_schema,
            python_schema=core_schema.union_schema(
                [
                    # check if it's an instance first before doing any further work
                    core_schema.is_instance_schema(PydanticObjectId),
                    from_str_schema,
                ]
            ),
            serialization=core_schema.plain_serializer_function_ser_schema(
                lambda instance: instance.x
            ),
        )
    @classmethod
    def __get_pydantic_json_schema__(
        cls, _core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
    ) -> JsonSchemaValue:
        # Use the same schema that would be used for `int`
        return handler(core_schema.int_schema())

我在 StackOverflow 上搜索了答案，但我找到的所有答案都涉及旧版本的 Pydantic，并使用与我上面粘贴的代码类似的代码。如果有人知道替代解决方案或可以提供有关如何在最新版本的 PyDantic 中定义自定义数据类型的明确指导，我将不胜感激。

---

更新

由于我没有正确创建 ObjectId 类型，我收到的一个持续错误是这样的

无法为 <class 'bson.objectid.ObjectId'> 生成 pydantic-core 架构。arbitrary_types_allowed=True在 model_config 中设置以忽略此错误或__get_pydantic_core_schema__在您的类型上实现以完全支持它。

如果您在调用 handler() 时遇到此错误，__get_pydantic_core_schema__那么您可能需要调用handler.generate_schema(<some type>)，因为我们不会调用__get_pydantic_core_schema__以<some type>避免无限递归。

有关更多信息，请访问https://errors.pydantic.dev/2.0.2/u/schema-for-unknown-type

而答案就是将其声明为未知类型，但我不想要它，我想将其声明为ObjectId。

Answer 1

一般来说，最好在 pydantic 的 GitHub 讨论中提出这样的问题。

你的解决方案非常接近，我认为你只是有错误的核心模式。

我认为我们关于使用自定义类型的文档Annotated很好地涵盖了这一点，但只是为了帮助您，这里有一个有效的实现：

from typing import Annotated, Any

from bson import ObjectId
from pydantic_core import core_schema

from pydantic import BaseModel

from pydantic.json_schema import JsonSchemaValue


class ObjectIdPydanticAnnotation:
    @classmethod
    def validate_object_id(cls, v: Any, handler) -> ObjectId:
        if isinstance(v, ObjectId):
            return v

        s = handler(v)
        if ObjectId.is_valid(s):
            return ObjectId(s)
        else:
            raise ValueError("Invalid ObjectId")

    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, _handler) -> core_schema.CoreSchema:
        assert source_type is ObjectId
        return core_schema.no_info_wrap_validator_function(
            cls.validate_object_id, 
            core_schema.str_schema(), 
            serialization=core_schema.to_string_ser_schema(),
        )

    @classmethod
    def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
        return handler(core_schema.str_schema())




class Model(BaseModel):
    id: Annotated[ObjectId, ObjectIdPydanticAnnotation]


print(Model(id='64b7abdecf2160b649ab6085'))
print(Model(id='64b7abdecf2160b649ab6085').model_dump_json())
print(Model(id=ObjectId()))
print(Model.model_json_schema())
print(Model(id='foobar'))  # will error