WARNING!!
This article is outdated.
Please, follow the current documentation of Beanie to use actual features and patterns.
Link to the doc - https://roman-right.github.io/beanie/
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
OR
poetry add beanie
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] = []
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])
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"])
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()
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
POST Input: Output:Click to see request details
localhost:10001/v1/notes
{
"title": "Monday",
"text": "Is the best day ever!"
}
{
"_id": "60425951ded355386e0666ed",
"title": "Monday",
"text": "Is the best day ever!",
"tag_list": []
}
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)
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
GET Output:Click to see request details
localhost:10001/v1/notes/60425951ded355386e0666ed
{
"_id": "60425951ded355386e0666ed",
"title": "Monday",
"text": "Is the best day ever!",
"tag_list": []
}
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()}}
)
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
PUT Input: Output:Click to see request details
localhost:10001/v1/notes/60425951ded355386e0666ed/add_tag
{
"name": "false",
"color": "RED"
}
{
"_id": "60425951ded355386e0666ed",
"title": "Monday",
"text": "Is the best day ever!",
"tag_list": [
{
"name": "false",
"color": "RED"
}
]
}
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()
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)
DELETE Output:Click to see request details
localhost:10001/v1/notes/60425951ded355386e0666ed
{
"status": "DELETED"
}
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()
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()
GET Output: GET Output:Click to see request details
localhost:10001/v1/notes
[
{
"_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"
}
]
}
]
localhost:10001/v1/notes/by_tag/true
[
{
"_id": "60425adeded355386e0666ef",
"title": "Monday",
"text": "Is, probably, not the best day ever..",
"tag_list": [
{
"name": "true",
"color": "GREEN"
}
]
}
]
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()
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()
GET Output:Click to see request details
localhost:10001/v1/notes/aggregate/by_tag_name
[
{
"_id": "false",
"total": 1
},
{
"_id": "true",
"total": 1
}
]
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:
Top comments (6)
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?
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:
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 :-)
Thank you for the response! I think that that's more than enough to get an idea.
Just wanted to say hi, and if you ever need anything or want to chat and meet the python team, ping me. :)
Hi Rachelle,
Nice! Thank you. I will text you. :-)
This is awesome :) FYI, we are planning to use beanie in a couple of our services in production. Also, we are using Sanic as the python asyncio based web framework. Thanks for this awesome ODM!