我正在尝试将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)
我在这里肯定缺少一些东西。
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 类。
皮丹蒂克 2
使用后验证器 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'))
另一种方法是使用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}
Getting Started with MongoDB and FastAPI
蒙戈开发者
此代码可帮助您使用 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
查看答案和其他文章，我使用以下对象并使用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
经过多次实验，我找到了这个解决方案：
使用 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
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")