本周,我开始使用MongoDB和Flask,所以我找到了一个关于如何使用它们的helpful article,通过使用PyDtic库来定义MongoDB的模型.然而,这篇文章有些过时了,大部分可以更新到新的PyDtic版本,但问题是对象ID是第三方字段,并且在不同版本之间发生了巨大的变化.

本文使用以下代码定义了对象ID:

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

过go ,这个代码运行得很好.然而,我最近发现,最新版本的PyDatics有一种更复杂的定义自定义数据类型的方法.我试过遵循Pydantic documentation,但我仍然感到困惑,一直未能成功实施.

我try 了这个实现来完成third party types个的实现,但它不起作用.这与文档中的代码几乎相同,只是字符串的int和对象ID的第三方调用标签有所不同.再说一次,我不确定为什么它不起作用.

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上搜索了答案,但我找到的所有答案都引用了早期版本的Pydtic,并使用了与我上面粘贴的代码类似的代码.如果有人知道替代解决方案,或者可以提供关于如何在最新版本的PyDatics中定义定制数据类型的明确指导,我将不胜感激.


更新

由于我没有创建正确的对象ID类型,我收到了一个持续的错误

无法为<类‘bson.objectid.ObjectId’>生成PYDANIC-CORE架构.在MODEL_CONFIG中设置arbitrary_types_allowed=True以忽略此错误,或在您的类型上实现__get_pydantic_core_schema__以完全支持它.

如果通过在__get_pydantic_core_schema__内调用Handler()得到此错误,则可能需要调用handler.generate_schema(<some type>),因为我们不会在<some type>上调用__get_pydantic_core_schema__,否则将避免无限递归.

欲了解更多信息,请访问https://errors.pydantic.dev/2.0.2/u/schema-for-unknown-type

答案是将其声明为未知类型,但我不想要它,我想将其声明为对象ID.

推荐答案

一般来说,最好是在PYDANTIC的GitHub讨论中提出这样的问题.

您的解决方案非常接近,我认为您只是使用了错误的核心架构.

我认为我们的documentation on using custom types via 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

Python相关问答推荐

Gekko解算器错误results.json未找到,无法找出原因

如何确保Flask应用程序管理面板中的项目具有单击删除功能?

修剪Python框架中的尾随NaN值

Python(Polars):使用之前的变量确定当前解决方案的Vector化操作

单击cookie按钮,但结果不一致

取相框中一列的第二位数字

Python中的Pool.starmap异常处理

在Python中管理多个OpenGVBO和VAO实例

比较两个数据帧并并排附加结果(获取性能警告)

由于NEP 50,向uint 8添加-256的代码是否会在numpy 2中失败?

SQLGory-file包FilField不允许提供自定义文件名,自动将文件保存为未命名

如何使用matplotlib在Python中使用规范化数据和原始t测试值创建组合热图?

为什么默认情况下所有Python类都是可调用的?

将tdqm与cx.Oracle查询集成

从嵌套的yaml创建一个嵌套字符串,后面跟着点

启用/禁用shiny 的自动重新加载

为什么Django管理页面和我的页面的其他CSS文件和图片都找不到?'

在Python 3中,如何让客户端打开一个套接字到服务器,发送一行JSON编码的数据,读回一行JSON编码的数据,然后继续?

可以bcrypts AES—256 GCM加密损坏ZIP文件吗?

numpy.unique如何消除重复列?