MongoDB ODM based on Pydantic and Motor
Project description
Beanie - is an asynchronous ODM for MongoDB, based on Motor and Pydantic.
It uses an abstraction over Pydantic models and Motor collections to work with the database. Class Document allows to create, replace, update, get, find and aggregate.
Here you can see, how to use Beanie, in simple examples:
Installation
PIP
pip install beanie
Poetry
poetry add beanie
Usage
Init
from typing import List
import motor
from pydantic import BaseSettings, BaseModel
from collections import Collection
from documents import Document
# CREATE BEANIE DOCUMENT STRUCTURE
class SubDocument(BaseModel):
test_str: str
class DocumentTestModel(Document):
test_int: int
test_list: List[SubDocument]
test_str: str
# CREATE MOTOR CLIENT AND DB
client = motor.motor_asyncio.AsyncIOMotorClient(
"mongodb://user:pass@host:27017/db",
serverSelectionTimeoutMS=100
)
db = client.beanie_db
# INIT BEANIE
init_beanie(database=db, document_models=[DocumentTestModel])
Create
Create a document (insert it)
document = DocumentTestModel(
test_int=42,
test_list=[SubDocument(test_str="foo"), SubDocument(test_str="bar")],
test_str="kipasa",
)
await document.create()
Insert one document
document = DocumentTestModel(
test_int=42,
test_list=[SubDocument(test_str="foo"), SubDocument(test_str="bar")],
test_str="kipasa",
)
await DocumentTestModel.insert_one(document)
Insert many documents
document_1 = DocumentTestModel(
test_int=42,
test_list=[SubDocument(test_str="foo"), SubDocument(test_str="bar")],
test_str="kipasa",
)
document_2 = DocumentTestModel(
test_int=42,
test_list=[SubDocument(test_str="foo"), SubDocument(test_str="bar")],
test_str="kipasa",
)
await DocumentTestModel.insert_many([document_1, document_2])
Find
Get the document
document = await DocumentTestModel.get(DOCUMENT_ID)
Find one document
document = await DocumentTestModel.find_one({"test_str": "kipasa"})
Find many documents
async for document in DocumentTestModel.find_many({"test_str": "uno"}):
print(document)
OR
documents = await DocumentTestModel.find_many({"test_str": "uno"}).to_list()
Find all the documents
async for document in DocumentTestModel.find_all()
print(document)
OR
documents = await DocumentTestModel.find_all().to_list()
Update
Replace the document (full update)
document.test_str = "REPLACED_VALUE"
await document.replace()
Replace one document
Replace one doc data by another
new_doc = DocumentTestModel(
test_int=0,
test_str='REPLACED_VALUE',
test_list=[]
)
await DocumentTestModel.replace_one({"_id": document.id}, new_doc)
Update the document (partial update)
in this example, I’ll add an item to the document’s “test_list” field
to_insert = SubDocument(test_str="test")
await document.update(update_query={"$push": {"test_list": to_insert.dict()}})
Update one document
await DocumentTestModel.update_one(
update_query={"$set": {"test_list.$.test_str": "foo_foo"}},
filter_query={"_id": document.id, "test_list.test_str": "foo"},
)
Update many documents
await DocumentTestModel.update_many(
update_query={"$set": {"test_str": "bar"}},
filter_query={"test_str": "foo"},
)
Update all the documents
await DocumentTestModel.update_all(
update_query={"$set": {"test_str": "bar"}}
)
Delete
Delete the document
await document.delete()
Delete one documents
await DocumentTestModel.delete_one({"test_str": "uno"})
Delete many documents
await DocumentTestModel.delete_many({"test_str": "dos"})
Delete all the documents
await DocumentTestModel.delete_all()
Aggregate
async for item in DocumentTestModel.aggregate(
[{"$group": {"_id": "$test_str", "total": {"$sum": "$test_int"}}}]
):
print(item)
OR
class OutputItem(BaseModel):
id: str = Field(None, alias="_id")
total: int
async for item in DocumentTestModel.aggregate(
[{"$group": {"_id": "$test_str", "total": {"$sum": "$test_int"}}}],
item_model=OutputModel
):
print(item)
OR
results = await DocumentTestModel.aggregate(
[{"$group": {"_id": "$test_str", "total": {"$sum": "$test_int"}}}],
item_model=OutputModel
).to_list()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
beanie-0.2.1.tar.gz
(10.6 kB
view hashes)
Built Distribution
beanie-0.2.1-py3-none-any.whl
(14.4 kB
view hashes)