使用Pydantic与FastAPI高效验证存储MongoDB ObjectID的方法

直接通过自定义Pydantic类型是最标准、高效的方案，既能自动完成字符串到bson.ObjectId的验证转换，又能避免手动实例化模型的冗余代码，直接在路径参数或请求体模型中使用即可。

1. 定义自定义ObjectId类型 #

封装验证逻辑和序列化规则，让Pydantic自动识别处理：

from bson import ObjectId
from pydantic import GetJsonSchemaHandler
from pydantic_core import core_schema

class PyObjectId(str):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetJsonSchemaHandler) -> core_schema.CoreSchema:
        # 定义核心验证规则：先按字符串解析，再转成ObjectId
        return core_schema.no_info_wrap_validator_function(
            cls.validate,
            core_schema.str_schema(),
            # 序列化时自动转回字符串，方便前端处理
            serialization=core_schema.plain_serializer_function_ser_schema(str),
        )

    @classmethod
    def validate(cls, value: str) -> ObjectId:
        try:
            return ObjectId(value)
        except Exception:
            raise ValueError("无效的ObjectId格式")

2. 在路径参数中直接使用 #

FastAPI会自动解析请求路径中的字符串，验证通过后直接返回ObjectId实例，无需手动转换：

from fastapi import FastAPI
import motor.motor_asyncio

app = FastAPI()
client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017")
db = client.test_db

@app.get("/users/{user_id}")
async def get_user(user_id: PyObjectId):
    # user_id已经是验证后的ObjectId，直接用于MongoDB查询
    user = await db.users.find_one({"_id": user_id})
    return user or {"message": "用户不存在"}

3. 在请求体模型中使用 #

如果需要从请求体接收ID，直接把字段类型设为PyObjectId即可：

from pydantic import BaseModel

class UserUpdateRequest(BaseModel):
    user_id: PyObjectId
    new_name: str

@app.put("/users/update")
async def update_user(request: UserUpdateRequest):
    # request.user_id已验证为ObjectId，直接用于数据库操作
    result = await db.users.update_one(
        {"_id": request.user_id},
        {"$set": {"name": request.new_name}}
    )
    return {"修改条数": result.modified_count}

方案优势 #

• 无冗余代码：不需要手动写body = MyClass(id=id)这类转换逻辑，参数/请求体自动完成验证和转换
• 标准合规：遵循Pydantic官方扩展类型的规范，代码可维护性强
• 双向兼容：自动处理字符串到ObjectId的验证，以及ObjectId到字符串的序列化，适配前端和数据库的不同需求

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