Using bson.ObjectId in Pydantic v2

Question

I found some examples on how to use ObjectId within BaseModel classes. Basically, this can be achieved by creating a Pydantic-friendly class as follows:

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

However, this seems to be for Pydantic v1, as this mechanism has been superseeded by the __get_pydantic_core_schema__ classmethod. However, I have been unable to achieve an equivalent solution with Pydantic v2. Is it possible? What validators do I need? I tried to refactor things but was unable to get anything usable.

Answer 1

None of the previous solutions worked for me when I want to use PyObjectId not only as type of a Pydantic model field but also for input parameters (path or query).

But the ObjectIdField class in the most recent release of the pydantic-mongo package seems to work fine with Pydantic v2 and exactly reproduces the behaviour which I was used to from the Pydantic v1 implementation.

If you do not want to install an additional dependency, you could also directly use the ObjectIdField implementation from the package sources to find at: https://github.com/jefersondaniel/pydantic-mongo/blob/f517e7161a8fb10002ef64881f092f6f84b40971/pydantic_mongo/fields.py

UPDATE for Pydantic v2.4

This approach went well on Pydantic 2.3.0, but on Pydantic 2.4.2 the OpenAPI schema generation fails with the following exception when trying to access /docs:

pydantic.errors.PydanticInvalidForJsonSchema: 
Cannot generate a JsonSchema for core_schema.PlainValidatorFunctionSchema 
({'type': 'no-info', 
'function': <bound method ObjectIdField.validate of <class 'ObjectIdField'>>})

I could solve this by excluding the validator function from the json_schema only. This is the full implementation, which works for me with Pydantic 2.4.2:

from typing import Any

from bson import ObjectId
from pydantic_core import core_schema

class PyObjectId(str):
    @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)
            ),
        )

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

        return ObjectId(value)

UPDATE to limit serialization to JSON

As pointed out in the comments, the solution above serializes ObjectId to str in both 'python' and 'json' mode. To limit this to 'json' mode, one could change the serialization parameter to:

serialization=core_schema.plain_serializer_function_ser_schema(
                lambda x: str(x),
                when_used='json'
            )