如何实现Pydantic模型中id字符串转ObjectId并映射为_id？

问题描述 #

用户定义了如下Pydantic模型：

from typing import Annotated
from bson import ObjectId

from pydantic import Field
from pydantic import EmailStr
from pydantic import BaseModel
from pydantic import BeforeValidator
from pydantic import ConfigDict
from pydantic import AwareDatetime
from pydantic import field_validator

# Represents an ObjectId field in the database.
# It will be represented as a `str` on the model so that it can
# be serialized to JSON.
PyObjectId = Annotated[str, BeforeValidator(str)]

class DBTableBase(BaseModel):
    # The primary key for the Table, stored as a `str` on the instance.
    # This will be aliased to `_id` when sent to MongoDB,
    # but provided as `id` in the API requests and responses.
    id: PyObjectId | None = Field(alias="_id",
                                  serialization_alias="id",
                                  default=None)
    model_config = ConfigDict(
        json_encoders={ObjectId: str},
        json_schema_extra={
            "example": {
                "id": "BSON_ID"
            }
        },
    )

class ClientModel(DBTableBase):
    first_name: str
    last_name: str

目前传入_id（ObjectId类型）创建ClientModel时，model_dump()能正常输出字符串类型的id。但想要实现反向逻辑：传入字符串类型的id创建对象时，自动转为ObjectId并映射到_id字段，可直接传id字符串后，model_dump()返回的id为None，需要解决这个问题。

解决方案 #

核心问题在于原模型的字段别名配置和数据转换逻辑不完整，需要调整PyObjectId的注解以及字段的验证、序列化规则：

1. 重构PyObjectId的类型注解 #

原PyObjectId只做了BeforeValidator(str)，现在需要添加字符串转ObjectId的验证器和ObjectId转字符串的序列化器，确保双向转换：

PyObjectId = Annotated[
    ObjectId,
    # 输入时：如果是合法字符串，转为ObjectId；否则保留原值
    BeforeValidator(lambda x: ObjectId(x) if isinstance(x, str) and ObjectId.is_valid(x) else x),
    # 序列化时：将ObjectId转为字符串，None则返回None
    PlainSerializer(lambda x: str(x) if x is not None else None)
]

2. 调整DBTableBase的字段配置 #

需要让模型同时接受id和_id作为输入字段，并正确映射到内部的_id（数据库字段），输出时统一用id字符串：

class DBTableBase(BaseModel):
    id: PyObjectId | None = Field(
        default=None,
        alias="_id",  # 数据库存储时的字段名
        validation_alias=AliasChoices("_id", "id"),  # 允许输入时用_id或id
        serialization_alias="id"  # 序列化输出时的字段名
    )
    model_config = ConfigDict(
        populate_by_name=True,  # 允许通过字段名（而非仅别名）填充数据
        json_schema_extra={
            "example": {
                "id": "60d21b4667d0d8992e610c85",
                "first_name": "John",
                "last_name": "Doe"
            }
        },
    )

3. 完整可运行代码 #

from typing import Annotated
from bson import ObjectId

from pydantic import Field, BaseModel, BeforeValidator, ConfigDict, PlainSerializer, AliasChoices

# 定义支持双向转换的PyObjectId类型
PyObjectId = Annotated[
    ObjectId,
    BeforeValidator(lambda x: ObjectId(x) if isinstance(x, str) and ObjectId.is_valid(x) else x),
    PlainSerializer(lambda x: str(x) if x is not None else None)
]

class DBTableBase(BaseModel):
    id: PyObjectId | None = Field(
        default=None,
        alias="_id",
        validation_alias=AliasChoices("_id", "id"),
        serialization_alias="id"
    )
    model_config = ConfigDict(
        populate_by_name=True,
        json_schema_extra={
            "example": {
                "id": "60d21b4667d0d8992e610c85",
                "first_name": "John",
                "last_name": "Doe"
            }
        },
    )

class ClientModel(DBTableBase):
    first_name: str
    last_name: str

4. 验证效果 #

• 传入字符串id创建对象：

client = ClientModel(id="60d21b4667d0d8992e610c85", first_name="John", last_name="Doe")
print(client.id)  # 输出：ObjectId('60d21b4667d0d8992e610c85')
print(client.model_dump())  # 输出：{'id': '60d21b4667d0d8992e610c85', 'first_name': 'John', 'last_name': 'Doe'}
print(client.model_dump(by_alias=True))  # 输出：{'_id': ObjectId('60d21b4667d0d8992e610c85'), 'first_name': 'John', 'last_name': 'Doe'}

• 传入_id（ObjectId类型）创建对象：

client2 = ClientModel(_id=ObjectId("60d21b4667d0d8992e610c85"), first_name="Jane", last_name="Smith")
print(client2.id)  # 输出：ObjectId('60d21b4667d0d8992e610c85')
print(client2.model_dump())  # 输出：{'id': '60d21b4667d0d8992e610c85', 'first_name': 'Jane', 'last_name': 'Smith'}

这样就实现了双向需求：输入字符串id自动转为ObjectId映射到_id，序列化输出字符串id；输入_id（ObjectId类型）也能正确处理，输出同样符合要求。

内容的提问来源于stack exchange，提问作者NPatel