Skip to content

Instantly share code, notes, and snippets.

@daaniam
Last active September 25, 2024 00:44
Show Gist options
  • Save daaniam/19ae60b5720a3e9774805f5dec5647ac to your computer and use it in GitHub Desktop.
Save daaniam/19ae60b5720a3e9774805f5dec5647ac to your computer and use it in GitHub Desktop.
MongoDB and Pydantic PyObjectId

Using MongoDB BSON ObjectId with Pydantic (FastAPI, pymongo or motor)

The important part here is when_used="json". This allows the serialization of data (ObjectId) as a string with FastAPI, while also ensuring that model_dump() retains the ObjectId for MongoDB. Not just for _id, but for every reference with ObjectId.

PyObjectId

from typing import Annotated, Any
from bson import ObjectId
from pydantic import Field
from pydantic_core import core_schema

__all__ = ["PyObjectId"]


class PyObjectId(str):
    """To create a pydantic Object that validates bson ObjectID"""

    @classmethod
    def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: Any) -> core_schema.CoreSchema:
        return core_schema.json_or_python_schema(
            json_schema=core_schema.str_schema(),
            python_schema=core_schema.union_schema(
                [
                    core_schema.is_instance_schema(ObjectId),
                    core_schema.chain_schema(
                        [
                            core_schema.str_schema(),
                            core_schema.no_info_plain_validator_function(cls.validate),
                        ]
                    ),
                ]
            ),
            serialization=core_schema.plain_serializer_function_ser_schema(lambda x: str(x), when_used="json"),
        )

    @classmethod
    def validate(cls, value) -> ObjectId:
        if not ObjectId.is_valid(value):
            raise ValueError("Invalid ObjectId")

        return ObjectId(value)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment