如何在 pydantic 模型中解析 ObjectId？

Posted 2023-02-19

【中文标题】如何在 pydantic 模型中解析 ObjectId？

【英文标题】：How to parse ObjectId in a pydantic model?

【发布时间】：2020-04-17 14:00:59

【问题描述】：

我正在尝试将 MongoDB 记录解析为 pydantic 模型，但对于 ObjectId 却没有这样做

据我了解，我需要为 ObjectId 设置验证器，并尝试扩展 ObjectId 类并使用 ObjectId 将 validator 装饰器添加到我的类中。我做了如下。

from pydantic import BaseModel, validator
from bson.objectid import ObjectId


class ObjectId(ObjectId):
    pass
    @classmethod
    def __get_validators__(cls):
        yield cls.validate
    @classmethod
    def validate(cls, v):
        if not isinstance(v, ObjectId):
            raise TypeError('ObjectId required')
        return str(v)


class User(BaseModel):
    who: ObjectId


class User1(BaseModel):
    who: ObjectId
    @validator('who')
    def validate(cls, v):
        if not isinstance(v, ObjectId):
            raise TypeError('ObjectId required')
        return str(v)

data = "who":ObjectId('123456781234567812345678')

不幸的是，两个“解决方案”都失败了：

>>> test = User(**data)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "pydantic/main.py", line 274, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for User
id
  field required (type=value_error.missing)

>>> test = User1(**data)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "pydantic/main.py", line 274, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for User1
who
  ObjectId required (type=type_error)

这里肯定有我遗漏的东西。

【参考方案1】：

您的第一个测试用例工作正常。问题在于你如何覆盖ObjectId。

from pydantic import BaseModel
from bson.objectid import ObjectId as BsonObjectId


class PydanticObjectId(BsonObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if not isinstance(v, BsonObjectId):
            raise TypeError('ObjectId required')
        return str(v)


class User(BaseModel):
    who: PydanticObjectId


print(User(who=BsonObjectId('123456781234567812345678')))

打印

who='123456781234567812345678'

只有 pydantic 应该使用 pydantic 类型。 Mongo 将为您提供 bsons ObjectId。所以用真实的 ObjectId 实例化你的数据。 所以data = "who":ObjectId('123456781234567812345678') 是错误的，因为它使用了您的子 ObjectId 类。

【参考方案2】：

另一种方法是使用 pydantic，我发现从另一个来源有用的是：

在模型文件夹中定义一个名为 PyObjectId.py 的文件。

from pydantic import BaseModel, Field as PydanticField
from bson import ObjectId

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")

然后你可以像这样在你的任何目标文件中使用它 users.py

from models.PyObjectId import PyObjectId
from pydantic import BaseModel, Field as PydanticField
from bson import ObjectId
class Users(BaseModel):
    id: PyObjectId = PydanticField(default_factory=PyObjectId, alias="_id")
    class Config:
        allow_population_by_field_name = True
        arbitrary_types_allowed = True #required for the _id 
        json_encoders = ObjectId: str

【参考方案3】：

这段代码帮助你使用json编码器

from bson import ObjectId
from pydantic import BaseModel


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

@classmethod
def validate(cls, v: str):
    try:
        return cls(v)
    except InvalidId:
        raise ValueError("Not a valid ObjectId")


class Foo(BaseModel):
    object_id_field: ObjId = None

    class Config:
        json_encoders = 
            ObjId: lambda v: str(v),
        



obj = Foo(object_id_field="60cd778664dc9f75f4aadec8")
print(obj.dict())
# 'object_id_field': ObjectId('60cd778664dc9f75f4aadec8')
print(obj.json())
# 'object_id_field': '60cd778664dc9f75f4aadec8'