I'm using FastAPI with MongoDB. I want my backend to respond to a simple get at domain/items/
with a list from the Mongodb database.
First, I extend the Mongodb ObjectId
class to be converted to string by FastAPI, and define my Item
model specifying in its config that ObjectId
and PyObjectId
types should be converted to string:
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")
class Item(BaseModel):
mongo_id: PyObjectId = Field(default_factory=PyObjectId, alias='_id')
name: str
class Config:
allow_population_by_field_name = True
json_encoders = {PyObjectId: str, ObjectId: str}
then, I define the get
method specifying the returned model:
@app.get("/items/", response_model=List[Item])
async def list_items(skip: int = 0, limit: int = 0):
"""List all items in the database"""
items = await ITEMS.find(skip=skip, limit=limit).to_list(MAX_TO_LIST)
return JSONResponse(status_code=status.HTTP_200_OK, content=items)
However, if I try to perform a GET request, an exception is raised from the line that returns the JSONResponse
:
TypeError: Object of type 'ObjectId' is not JSON serializable
First of all, I do not understand what is the difference between the json_encoders = {PyObjectId: str, ObjectId: str}
in the Item
model config and the field_schema.update(type="string")
in the PyObjectId
__modify_schema__()
method. Do we need both? And why?
Second, I do not understand why isn't the ObjectId
field of each item transformed into string automatically. What am I missing or doing wrong?
NOTE: I know I could just iterate the items
returned by Mongodb, transforming them to dict and trasnforming their '_id'
field into a string, but I would like FastAPI and Pydantic to do this automatically.
I solved it like this. First I declared this ObjectID class.
"""Defines the object id model/class."""
# ruff: noqa
from typing import Any
from bson import ObjectId
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import core_schema
class ObjectIdPydanticAnnotation:
"""Defines a wrapper class around the mongodb ObjectID class adding serialization."""
@classmethod
def validate_object_id(cls, v: Any, handler) -> ObjectId:
if isinstance(v, ObjectId):
return v
s = handler(v)
if ObjectId.is_valid(s):
return ObjectId(s)
else:
msg = "Invalid ObjectId"
raise ValueError(msg)
@classmethod
def __get_pydantic_core_schema__(
cls,
source_type,
_handler,
) -> core_schema.CoreSchema:
assert source_type is ObjectId # noqa: S101
return core_schema.no_info_wrap_validator_function(
cls.validate_object_id,
core_schema.str_schema(),
serialization=core_schema.to_string_ser_schema(),
)
@classmethod
def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
return handler(core_schema.str_schema())
And then I use it like this:
"""Defines the models/classes for rooms."""
from typing import Annotated
from bson import ObjectId
from models.object_id import ObjectIdPydanticAnnotation
from pydantic import BaseModel, Field
class Item(BaseModel):
"""Defines the Room class according to the database schema."""
id: Annotated[ObjectId, ObjectIdPydanticAnnotation] | None = Field(
default=None,
alias="_id",
)
name: str