Skip to main content

MongoDB ODM based on Pydantic and Motor

Project description

https://raw.githubusercontent.com/roman-right/beanie/main/assets/logo/with_text.svg

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 beanie import Document
from pydantic import BaseModel


# 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.3.tar.gz (10.6 kB view hashes)

Uploaded Source

Built Distribution

beanie-0.2.3-py3-none-any.whl (14.4 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page