DEV Community

loading...
Cover image for Announcing Beanie - MongoDB ODM

Announcing Beanie - MongoDB ODM

Roman
I love to code
Updated on ・6 min read

I'm excited to introduce Beanie - Python micro ODM (Object Document Mapper) for MongoDB!

The main component of Beanie is Pydantic - a popular library for data parsing and validation. It helps to implement the main feature - data structuring. Beanie Document - is an abstraction over the Pydantic BaseModel that allows working with Python objects at the application level and JSON objects at the database level. In the general case, one MongoDB collection is associated with one Beanie Document. This brings predictability when working with the database, and at the same time preserves all the flexibility of MongoDB documents - it is possible to represent any data structure with the Pydantic model (or even a group of structures with Optional and Union annotations).

I'm doing quite a few pet projects: Experiments with the new technologies and Proofs of Concepts. For these purposes, I needed a tool for working with the database that I can use immediately without a long setup. And with which I could change the data structure frequently, adding and dropping elements here and there. This is how Beanie was born.

Usage example

But that's a bit boring, isn't it? Now let's get to the interesting part - the usage examples. It will show how handy this tool is. I'm deliberately omitting some additional imports and helpers here, so as not to overload the picture and focus only on the important things. The entire working app from this article is in my GitHub repo beanie-fastapi-demo.

As an example, I'll create a small rest-service for managing notes.

Installation

pip install beanie
Enter fullscreen mode Exit fullscreen mode

OR

poetry add beanie
Enter fullscreen mode Exit fullscreen mode

Data model

The package is installed. Now we're ready to go. Let's define the structure of the notes.

from enum import Enum
from typing import Optional, List

from beanie import Document
from pydantic import BaseModel


class TagColors(str, Enum):
    RED = "RED"
    BLUE = "BLUE"
    GREEN = "GREEN"


class Tag(BaseModel):
    name: str
    color: TagColors = TagColors.BLUE


class Note(Document):  # This is the document structure
    title: str
    text: Optional[str]
    tag_list: List[Tag] = []
Enter fullscreen mode Exit fullscreen mode

A note consists of the required title, optional text, and a list of tags. Each tag has a name and a color. The class Note has implemented all this in a pydantic way.

Now I will create the database connection and Beanie initialization:

import motor.motor_asyncio
from beanie.general import init_beanie


async def main():
    # Create Motor client
    client = motor.motor_asyncio.AsyncIOMotorClient(
        f"mongodb://user:pass@host:27017/beanie_db"
    )

    # Init Beanie
    await init_beanie(client.beanie_db, document_models=[Note])
Enter fullscreen mode Exit fullscreen mode

No surprises here. Beanie uses Motor as an asynchronous driver for MongoDB. For initialization, I need to provide the list of all Beanie documents that I will be working with.

Web application

As API framework I will use the popular FastApi.

from fastapi import FastAPI

app = FastAPI()
app.include_router(notes_router, prefix="/v1", tags=["notes"])
Enter fullscreen mode Exit fullscreen mode

I will first implement a simple CRUD to show the basics of Beanie

Create

Before any endpoint implementation, I would like to show how to create a document:

note = Note(title="Monday", text="What a nice day!")
await note.create()
Enter fullscreen mode Exit fullscreen mode

The create method stores the document in the database. Also, Beanie allows document insertion in a few other ways, including batch insert. Examples of usage can be found in the Document method descriptions by the link

Now I'll demonstrate the same trick, but this time inside the endpoint:

from fastapi import APIRouter

notes_router = APIRouter()


@notes_router.post("/notes/", response_model=Note)
async def create_note(note: Note):
    # Note creation
    await note.create()

    return note
Enter fullscreen mode Exit fullscreen mode

Click to see request details

POST localhost:10001/v1/notes

Input:

{
  "title": "Monday",
  "text": "Is the best day ever!"
}
Enter fullscreen mode Exit fullscreen mode

Output:

{
  "_id": "60425951ded355386e0666ed",
  "title": "Monday",
  "text": "Is the best day ever!",
  "tag_list": []
}
Enter fullscreen mode Exit fullscreen mode

FastAPI uses Pydantic models to parse the request body. This means that I can use Beanie Document as the model and then work with the already parsed document. To insert it into the database, I use the create method again.

Read

In the response, it returns _id - the unique id of the document in the database. Now I'll show how to retrieve the note based on its id.

Independent implementation:

note = await Note.get(note_id)
Enter fullscreen mode Exit fullscreen mode

Inside the endpoint:

from beanie.fields import PydanticObjectId


# Helper method to get instances
async def get_note(note_id: PydanticObjectId) -> Note:
    # Note retrieval
    note = await Note.get(note_id)

    if note is None:
        raise HTTPException(
            status_code=404,
            detail="Note not found"
        )
    return note


# Actual endpoint
@notes_router.get("/notes/{note_id}", response_model=Note)
async def get_note_by_id(
        # Helper usage with Depends annotation
        note: Note = Depends(get_note)
):
    return note
Enter fullscreen mode Exit fullscreen mode

Click to see request details

GET localhost:10001/v1/notes/60425951ded355386e0666ed

Output:

{
  "_id": "60425951ded355386e0666ed",
  "title": "Monday",
  "text": "Is the best day ever!",
  "tag_list": []
}
Enter fullscreen mode Exit fullscreen mode

Update

The application can already create and read the notes, but it can't do anything with the tags yet. That's what I'm going to do next.

Independent implementation:

tag = Tag(name="false", color="RED")
await note.update(
    update_query={"$push": {"tag_list": tag.dict()}}
)
Enter fullscreen mode Exit fullscreen mode

Inside the endpoint:

@notes_router.put("/notes/{note_id}/add_tag", response_model=Note)
async def add_tag(tag: Tag, note: Note = Depends(get_note)):
    # Update the note
    await note.update(
        update_query={"$push": {"tag_list": tag.dict()}}
    )

    return note
Enter fullscreen mode Exit fullscreen mode

Click to see request details

PUT localhost:10001/v1/notes/60425951ded355386e0666ed/add_tag

Input:

{
  "name": "false",
  "color": "RED"
}
Enter fullscreen mode Exit fullscreen mode

Output:

{
  "_id": "60425951ded355386e0666ed",
  "title": "Monday",
  "text": "Is the best day ever!",
  "tag_list": [
    {
      "name": "false",
      "color": "RED"
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode

There are two main types of Beanie Document update:

  • replace - full update of the document
  • update - partial update of the document

Replace is useful in many cases, but in the current one, I don't know if the Note document is in the actual state now. Some tags might have been added after the last synchronization with the database. If I replace the document with new data, I can easily lose some data. That's why I use the partial update here. As argument, the update method takes a query in PyMongo query format.

Delete

The delete operation is not that interesting to talk much about.

Independent implementation:

await note.delete()
Enter fullscreen mode Exit fullscreen mode

Inside the endpoint:

@notes_router.delete("/notes/{note_id}", response_model=StatusModel)
async def get_note_by_id(note: Note = Depends(get_note)):
    # Delete the note
    await note.delete()

    return StatusModel(status=Statuses.DELETED)
Enter fullscreen mode Exit fullscreen mode

Click to see request details

DELETE localhost:10001/v1/notes/60425951ded355386e0666ed

Output:

{
  "status": "DELETED"
}
Enter fullscreen mode Exit fullscreen mode

Lists

CRUD is done, but no service gets around list endpoints. The implementation is simple again.

Independent implementation:

all_notes = await Note.find_all().to_list()
red_notes = Note.find_many({"tag_list.color": "RED"}).to_list()
Enter fullscreen mode Exit fullscreen mode

Inside the endpoint:

@notes_router.get(
    "/notes/",
    response_model=List[Note]
)
async def get_all_notes():
    # Get all notes
    return await Note.find_all().to_list()


@notes_router.get(
    "/notes/by_tag/{tag_name}",
    response_model=List[Note]
)
async def filter_notes_by_tag(tag_name: str):
    # Filter notes
    return await Note.find_many(
        {"tag_list.name": tag_name}
    ).to_list()
Enter fullscreen mode Exit fullscreen mode

Click to see request details

GET localhost:10001/v1/notes

Output:

[
  {
    "_id": "60425ac0ded355386e0666ee",
    "title": "Monday",
    "text": "Is the best day ever!",
    "tag_list": [
      {
        "name": "false",
        "color": "RED"
      }
    ]
  },
  {
    "_id": "60425adeded355386e0666ef",
    "title": "Monday",
    "text": "Is, probably, not the best day ever..",
    "tag_list": [
      {
        "name": "true",
        "color": "GREEN"
      }
    ]
  }
]
Enter fullscreen mode Exit fullscreen mode

GET localhost:10001/v1/notes/by_tag/true

Output:

[
  {
    "_id": "60425adeded355386e0666ef",
    "title": "Monday",
    "text": "Is, probably, not the best day ever..",
    "tag_list": [
      {
        "name": "true",
        "color": "GREEN"
      }
    ]
  }
]
Enter fullscreen mode Exit fullscreen mode

The find_all method tells everything about itself by name only. find_many is also simple. It takes PyMongo's query as an argument to filter the documents.

Aggregations

And finally, I want to show how to create aggregations with Beanie. In this example, I'll calculate how many notes I have per tag name.

Independent implementation:

class AggregationResponseItem(BaseModel):
    id: str = Field(None, alias="_id")
    total: int


results = await Note.aggregate(
    aggregation_query=[
        {"$unwind": "$tag_list"},
        {"$group": {
            "_id": "$tag_list.name",
            "total": {"$sum": 1}
        }}
    ],
    item_model=AggregationResponseItem
).to_list()
Enter fullscreen mode Exit fullscreen mode

Inside the endpoint:

@notes_router.get("/notes/aggregate/by_tag_name", response_model=List[AggregationResponseItem])
async def filter_notes_by_tag_name():
    # Notes aggregation
    return await Note.aggregate(
        aggregation_query=[
            {"$unwind": "$tag_list"},
            {"$group": {
                "_id": "$tag_list.name",
                "total": {"$sum": 1}
            }}
        ],
        item_model=AggregationResponseItem
    ).to_list()
Enter fullscreen mode Exit fullscreen mode

Click to see request details

GET localhost:10001/v1/notes/aggregate/by_tag_name

Output:

[
  {
    "_id": "false",
    "total": 1
  },
  {
    "_id": "true",
    "total": 1
  }
]
Enter fullscreen mode Exit fullscreen mode

In all the examples before aggregation, the result of the Note methods were Note objects or lists of the Noteobjects. But in the aggregation case, the result can have any structure. To continue work with the python objects I provide parameter item_model=AggregationResponseItem to the aggregate method and it returns a list of the AggregationResponseItem objects.

Conclusion

Well, the service is done. I've shown how easy it is to do things with Beanie. You can stop thinking about parsing and validating database data and just focus on your project. Certainly, I didn't use all the possible functions. You can find the whole list of methods in the project description here

Beanie has become a very important part of my development toolset. Especially when it comes to building prototypes. And I'm happy to share it with the community. I continue to use it in my projects - which means I continue to develop it. The next big thing I want to implement is structure and data migrations.

You are always welcome to participate in the development :-) Thank you very much for your time!

GitHub and PyPI project links:

Demo project from this article:

Discussion (5)

Collapse
denpetrov profile image
denpetrov

Great article!

I know that your post is not about comparison, but did you look at MongoEngine? Maybe you had a chance to compare them? Or what do you think about the differences at first glance?

Collapse
romanright profile image
Roman Author • Edited

Hello,
Thank you!

This is a complex topic. I think it is hard to compare these tools because it is on the different stages of the development, use different approaches and, I'd say, should be used in different cases.

The main differences are:

  • MongoEngine is a complete ODM with many and many features inside. Beanie is a micro ODM that supports very basic operations for now.
  • MongoEngine uses its own field types to parse and validate data. Beanie (thanks to Pydantic) uses Python types and Pydantic custom fields for that.
  • MongoEngine is synchronous. Beanie uses an async engine.

To be honest, I have no plan to make the full comparison in the future too. The results will not be fair, or will not look fair, because I'm the author of one of the competitors. Even if I'll try hard to avoid it :-)

Collapse
denpetrov profile image
denpetrov

Thank you for the response! I think that that's more than enough to get an idea.

Collapse
techbelle profile image
rachelle palmer

Just wanted to say hi, and if you ever need anything or want to chat and meet the python team, ping me. :)

Collapse
romanright profile image
Roman Author

Hi Rachelle,
Nice! Thank you. I will text you. :-)

Forem Open with the Forem app