MongoDB / Beanie Integration

[rgajason]

Is anyone using MongoDB as the database with Starlite? Would like to hear how you went about integrating. Also would love to hear from anyone using Beanie as the ODM.

[rgajason]

I've started in on this myself using Beanie, but am running into some interesting behavior. I'm coming from FastAPI, so hopefully I'm just misunderstanding some differences between the frameworks.

I initialize Beanie using on_startup:

async def initialize_beanie(state: State) -> None:
    settings = get_settings()
    cx = AsyncIOMotorClient(settings.mongodb_url, uuidRepresentation="standard")
    db = getattr(cx, settings.mongodb_db)
    await init_beanie(
        database=db,
        document_models=[
            Widget,
        ],
        allow_index_dropping=True,
    )

app = Starlite(
    route_handlers=[root, SandboxController],
    on_startup=[initialize_beanie],
)

I define a model like so:

from beanie import Document, Indexed

class Widget(Document):
    name: Indexed(str, unique=True)

    class Settings:  # Beanie configuration
        name = "sandbox-widgets"

    class Config:  # Pydantic configuration
        schema_extra = {
            "example": {
                "name": "Widget Name Goes Here",
            },
        }

Creating new instances is where it gets a little wonky. This code creates the document in the DB, but returns an HTTP 500 with a message of "Unable to serialize response content":

@post(
    "/widget",
    response_description="Add a new Sandbox Widget",
)
async def create_widget(self, data: Widget) -> Widget:
    try:
        await data.create()
    except DuplicateKeyError as e:
        raise DuplicateKeyIndexException(e, "widget")
    return data

If I say return data.json() instead of return data it seems to work:

@post(
    "/parent",
    response_description="Add a new Sandbox Widget",
)
async def create_widget(self, data: Widget) -> Widget:
    try:
        await data.create()
    except DuplicateKeyError as e:
        raise DuplicateKeyIndexException(e, "widget")
    return data.json()

Any thoughts on the current approach?

[Goldziher]

So, for some rason its failing to serialize the Widget instance as a pydantic class. This should not be happening. I will take a look and get back to you.

[Goldziher]

ok, i've found the issue.

Our serializer has the following logic:

def default_serializer(value: Any) -> Any:
    """Return the default serializer for a given object based on its type.

    Args:
        value: A value to serialize
    Returns:
        A serialized value
    Raises:
        TypeError: if value is not supported
    """
    if isinstance(value, BaseModel):
        return value.dict()
    if isinstance(value, SecretStr):
        return value.get_secret_value()
    if isinstance(value, (PurePath, PurePosixPath)):
        return str(value)
    raise TypeError("unsupported type")

When we test if isinstance(value, BaseModel), it returns true. But when we call value.dict(), it returns the following value:

{'id': ObjectId('637e5eb67e4586cddb7cfa21'), 'name': 'moishe zuchmir'}

If I inspect id, what I see is this:

This value IS NOT json serializable as such. The reason you can use this with fastapi is that it relies on the models buildin JSON method, which we do not.

Options on your end:

1. define a custom response serializer that handles the Document type from beanie: https://starlite-api.github.io/starlite/usage/5-responses/10-custom-responses/?h=custom
2. open a bug issue on beanie and make sure this is fixed there.
3. do both of the above.

[rgajason]

As a workaround I created the following custom response and added it to my controllers:

Edit: updated to support references (Links)

from typing import Any, Dict

from beanie import Link, PydanticObjectId
from starlite import Response


class BeanieResponse(Response):
    def serializer(self, value: Any) -> Dict[str, Any]:
        if isinstance(value, PydanticObjectId):
            return str(value)
        if isinstance(value, Link):
            return value.to_dict().get("id")
        return super().serializer(value)

I'm not quite following why that is working because I'm not returning a dict, but it is...

First I tried:

from typing import Any, Dict

from beanie import Document
from starlite import Response


class BeanieResponse(Response):
    def serializer(self, value: Any) -> Dict[str, Any]:
        if isinstance(value, Document):
            return value.dict()
        return super().serializer(value)

...but I would get the same error. Through debugging I could see that the serializer was being called multiple times - once for the document and again for individual fields. The document serialization would work but not the ID field.

[Goldziher]

its working because orjson is calling the serilizer function for each type of value it is not natively capable of handling.

[bwhli]

As a workaround I created the following custom response and added it to my controllers:

Edit: updated to support references (Links)

from typing import Any, Dict

from beanie import Link, PydanticObjectId
from starlite import Response


class BeanieResponse(Response):
    def serializer(self, value: Any) -> Dict[str, Any]:
        if isinstance(value, PydanticObjectId):
            return str(value)
        if isinstance(value, Link):
            return value.to_dict().get("id")
        return super().serializer(value)

I'm not quite following why that is working because I'm not returning a dict, but it is...

First I tried:

from typing import Any, Dict

from beanie import Document
from starlite import Response


class BeanieResponse(Response):
    def serializer(self, value: Any) -> Dict[str, Any]:
        if isinstance(value, Document):
            return value.dict()
        return super().serializer(value)

...but I would get the same error. Through debugging I could see that the serializer was being called multiple times - once for the document and again for individual fields. The document serialization would work but not the ID field.

Beanie also supports bson.int64.Int64 type, so I had to add another check like so:

class BeanieResponse(Response):
    def serializer(self, value: Any) -> Dict[str, Any]:
        if isinstance(value, PydanticObjectId):
            return str(value)
        if isinstance(value, Link):
            return value.to_dict().get("id")
        if isinstance(value, Int64):
            return int(value)
        return super().serializer(value)

[rgajason]

Nice addition - thanks for sharing!

[bwhli]

FYI, I raised an issue on the Beanie repo and the author will look into it and fix it. I'll do some testing on Starlite after the issue is fixed in Beanie.

BeanieODM/beanie#458

[roman-right]

Hi there,
There is no way to tell from inside the class, how this class should be serialized by json package. The encoding process here is out of custom types' control. Documentation of json suggests using custom encoders.

[provinzkraut]

This can be either solved by using type_encoders to tell Starlite how to serialize beanie Documents, or by creating a Plugin.

If all you need is serialization into JSON, type_encoders are the way to go. The simplest solution would be something like this:

from msgspec import Raw
from beanie import Document
from starlite import Starlite

app = Starlite([...], type_encoders={Document: lambda d: Raw(d.json().encode("utf-8"))})

This way you can make use of beanies built-in json capabilities.

[jcrist]

In the upcoming msgspec release (hoping out this week) this can be simplified slightly to drop the .encode("utf-8") call. For realistic message sizes this is mostly an ergonomic benefit, but it does shave off a fraction of the conversion time. See jcrist/msgspec#252.

[rgajason]

For Starlite >= 1.48.0, see #1072