如何在 pydantic 模型中解析 ObjectId？

Question

我正在尝试将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')}

很遗憾，两种“解决方案”均存在以下问题：

Translated text:

Unfortunately, both "solutions" have the following problems:

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

我肯定在这里缺少了某些东西。

- roshii

Answer 1

Pydantic 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类。

Pydantic 2

使用AfterValidator https://docs.pydantic.dev/latest/usage/validators/

from typing_extensions import Annotated
from pydantic import BaseModel
from pydantic.functional_validators import AfterValidator
from bson import ObjectId as _ObjectId


def check_object_id(value: str) -> str:
    if not _ObjectId.is_valid(value):
        raise ValueError('Invalid ObjectId')
    return value


ObjectId = Annotated[str, AfterValidator(check_object_id)]


class Example(BaseModel):
    id: ObjectId


print(Example(id='5f9b3b3b9d9f3d0001a3b3b3'))
print(Example(id='1'))

- Tom Wojcik

---

Pydantic2的新解决方案是什么？ - Ali Rn

@AliRn https://docs.pydantic.dev/latest/usage/validators/我做了这个from typing_extensions import Annotated from pydantic import BaseModel from pydantic.functional_validators import AfterValidator from bson import ObjectId as _ObjectId def check_object_id(value: str) -> str: if not _ObjectId.is_valid(value): raise ValueError('Invalid ObjectId') return value ObjectId = Annotated[str, AfterValidator(check_object_id)] class Example(BaseModel): id: ObjectId print(Example(id='5f9b3b3b9d9f3d0001a3b3b3')) print(Example(id='1')) - Heichou

Answer 2

我发现另一种有用的方法是使用pydantic：

在models文件夹中定义一个名为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}

- Mohammed

Answer 3

开始使用MongoDB和FastAPI

Mongo开发者

这段代码可以帮助您使用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'}

更新：

你可以在你的 pydantic 模型中使用这个字段类型：

from bson import ObjectId as BaseObjectId

class ObjectId(str):
"""Creating a ObjectId class for pydantic models."""

    @classmethod
    def validate(cls, value):
        """Validate given str value to check if good for being ObjectId."""
        try:
            return BaseObjectId(str(value))
        except InvalidId as e:
            raise ValueError("Not a valid ObjectId") from e

    @classmethod
    def __get_validators__(cls):
        yield cls.validate

- milad_vayani

Answer 4

在查找答案和其他文章时，我使用以下对象，并使用pydantic.json中的ENCODERS_BY_TYPE将编码从str全局变为ObjectId，反之亦然。

import bson
import bson.errors 
from pydantic.json import ENCODERS_BY_TYPE


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

    @classmethod
    def validate(cls, v):
        raise_error = False

        try:
            if isinstance(v, str):
                v = bson.ObjectId(v)

            if (
                not isinstance(v, (bson.ObjectId, cls))
                or not bson.ObjectId.is_valid(v)
            ):
                raise_error = True
        except bson.errors.InvalidId:
            raise_error = True

        if raise_error:
            raise ValueError("Invalid ObjectId")

        return v

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


if ObjectId not in ENCODERS_BY_TYPE:
    ENCODERS_BY_TYPE[ObjectId] = str
    ENCODERS_BY_TYPE[bson.ObjectId] = str

- Felipe Buccioni

Answer 5

经过多次尝试，我找到了这个解决方案：

已在Python 3.11上测试

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


@classmethod
def __get_validators__(cls):
    yield injected_validator


def injected_validator(v):
    if not isinstance(v, ObjectId):
        raise TypeError('ObjectId required')

    return v


# This does the trick. It forces ObjectId to have a validator 
ObjectId.__get_validators__ = __get_validators__


def parse_object_id(v):
    if isinstance(v, str) and ObjectId.is_valid(v):
        return ObjectId(v)    

    if isinstance(v, ObjectId):
        return v

    raise TypeError(f"Invalid ObjectId: {v}")


class MyModel(BaseModel):
    id: ObjectId | None

    @validator("id", pre=True)
    def ensure_id_is_object_id(cls, v):
        return None if v is None else parse_object_id(v)


def ensure_oid(v):
    assert type(v.id) == ObjectId


assert MyModel().id is None

ensure_oid(MyModel(id=ObjectId()))
ensure_oid(MyModel(id=ObjectId("642796132887d08ca3a7a986")))

# Intellisense warn (but works): Expected type 'ObjectId | None', got 'str' instead
ensure_oid(MyModel(id="642796430b2fb0ed6292d1d2"))

ensure_oid(MyModel.parse_obj({"id": ObjectId()}))
ensure_oid(MyModel.parse_obj({"id": "642796893cd44d9ff690a455"}))
ensure_oid(MyModel.parse_obj({"id": ObjectId("642796abb14eb1e6a9183ae5")}))
ensure_oid(MyModel.parse_raw('{"id": "642796924f9a0adbea020d60"}'))

很遗憾，我无法让_id字段名称正常工作。 如果您找到解决方案，请与我分享！

解决此问题的方法是创建类似于以下内容的属性_id:

@property
def _id(self) -> ObjectId | None:
    return self.id

- Michael Pacheco

Answer 6

我将分享关于 pydantic 2 的解决方案。

根据我的使用情况，当数据以 ObjectId 形式进入模型时，我需要将其解析为字符串。当数据以字符串形式进入时，我需要将其解析为 ObjectId。

from typing_extensions import Annotated

from pydantic import BaseModel, ConfigDict
from pydantic.functional_validators import AfterValidator

from bson.objectid import ObjectId


def object_id_validate(v: ObjectId | str) -> ObjectId | str:
    assert ObjectId.is_valid(v), f'{v} is not a valid ObjectId'
    if isinstance(v, str):
        return ObjectId(v)
    return str(v)


PyObjectId = Annotated[ObjectId | str, AfterValidator(object_id_validate)]


class MyModel(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    user_id: PyObjectId


print(MyModel(user_id=str(ObjectId()))) # user_id=ObjectId('653087c8c8640ef5700a1bb5')
print(MyModel(user_id=ObjectId())) # user_id='653087c8c8640ef5700a1bb6'

- Anderson Araújo

Answer 7

Tom Wojcik的解决方案稍作修改后对我有用：

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 Bird(BaseModel):
    id: PydanticObjectId = Field(..., alias="_id")

- Vlad Karassev