Pydantic-based Mongo ORM
Project description
Description
Beanie - is asynchronous ORM for MongoDB, based on Motor and Pydantic.
Beanie uses an abstraction over Pydantic Models Motor collections to work with mongo. Document and Collection classes allow to create, replace, update, get, find and aggregate.
One collection can be associated with only one Document and it helps to keep it structured.
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 MOTOR CLIENT AND DB
client = motor.motor_asyncio.AsyncIOMotorClient(
"mongodb://user:pass@host:27017/db",
serverSelectionTimeoutMS=100
)
db = client.beanie_db
# CREATE BEANIE DOCUMENT STRUCTURE
class SubDocument(BaseModel):
test_str: str
class DocumentTestModel(Document):
test_int: int
test_list: List[SubDocument]
test_str: str
# CREATE BEANIE COLLECTION WITH DocumentTestModel STRUCTURE
test_collection = Collection(
name="test_collection", db=db, document_model=DocumentTestModel
)
Documents
Create a document (Insert)
document = DocumentTestModel(
test_int=42,
test_list=[SubDocument(test_str="foo"), SubDocument(test_str="bar")],
test_str="kipasa",
)
await document.create()
Replace the document (full update)
document.test_str = "REPLACED_VALUE"
await document.replace()
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()}})
Get the document
document = await DocumentTestModel.get(DOCUMENT_ID)
Find one document
document = await DocumentTestModel.find_one({"test_str": "kipasa"})
Find the documents
async for document in DocumentTestModel.find({"test_str": "uno"}):
print(document)
OR
documents = await DocumentTestModel.find({"test_str": "uno"}).to_list()
Get all the documents
async for document in DocumentTestModel.all()
print(document)
OR
documents = await DocumentTestModel.all().to_list()
Delete the document
await document.delete()
Aggregate from the document model
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()
Collections
Insert the document into the collection
inserted_document = await collection.insert_one(document)
Replace the document
await collection.replace_one(document)
Update the document
to_insert = SubDocument(test_str="test")
await collection.update_one(
document, update_query={"$push": {"test_list": to_insert.dict()}}
)
Update many documents
await collection.update_many(
update_query={"$set": {"test_int": 100}}, filter_query={"test_str": "kipasa"},
)
Delete the document
await collection.delete_one(document)
Get the document
document = await collection.get_one(DOCUMENT_ID)
Find the document
document = await collection.find_one({"test_str": "one"})
Find many documents
async for document in collection.find({"test_str": "uno"}):
print(document)
OR
documents = await collection.find({"test_str": "uno"}).to_list()
OR
documents = await collection.find({"test_str": "uno"}).to_list(length=10)
Get all the documents from the collection
async for document in collection.all():
print(document)
OR
documents = await collection.all().to_list()
Aggregate
async for item in collection.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 collection.aggregate(
[{"$group": {"_id": "$test_str", "total": {"$sum": "$test_int"}}}],
item_model=OutputModel
):
print(item)
OR
results = await collection.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.1.0.tar.gz
(11.3 kB
view hashes)
Built Distribution
beanie-0.1.0-py3-none-any.whl
(14.9 kB
view hashes)