motormongo 0.1.4

An asynchronous Object Document Mapper (O.D.M) for MongoDB built on-top of Motor.

Author: Patrick Prunty.

motormongo - An Object Document Mapper for MongoDB built on-top of Motor, the MongoDB recommended asynchronous Python driver for MongoDB Python applications, designed to work with Tornado or asyncio and enable non-blocking access to MongoDB.

Asynchronous operations in a backend system, built using FastAPI for example, enhances performance and scalability by enabling non-blocking, concurrent handling of multiple requests, leading to more efficient use of server resources.

The interface for instantiating Document classes follows similar logic to mongoengine, enabling ease-of-transition and migration from mongoengine to motormongo.

Installation

To install motormongo, you can use pip inside your virtual environment:

python -m pip install motormongo

Alternatively, to install motormongo into your poetry environment:

poetry add motormongo

Quickstart

Step 1. Create a motormongo client:

import asyncio
from motormongo import DataBase

async def init_db():
    # This 'connect' method needs to be called inside of an async function
    await DataBase.connect(uri="<mongo_uri>", database="<mongo_database>")

if __name__ == "__main__":
    asyncio.run(init_db())

or, in a FastAPI application:

import os
from fastapi import FastAPI
from motormongo import DataBase

app = FastAPI()

@app.on_event("startup")
async def startup_db_client():
    await DataBase.connect(uri=os.getenv("MONGODB_URL"), db=os.getenv("MONGODB_DB"))

@app.on_event("shutdown")
async def shutdown_db_client():
    await DataBase.close()

The mongo_uri should look something like this:

mongodb+srv://<username>:<password>@<cluster>.mongodb.net

and database should be the name of an existing MongoDB database in your MongoDB instance.

For details on how to set up a local or cloud MongoDB database instance, see here.

Step 2. Define a motormongo Document:

Define a motormongo User document:

import re
import bcrypt
from motormongo.abstracts.document import Document
from motormongo.fields.binary_field import BinaryField
from motormongo.fields.string_field import StringField
from motormongo.fields.integer_field import IntegerField
from motormongo.fields.enum_field import EnumField

def hash_password(password) -> bytes:
    # Example hashing function
    return bcrypt.hashpw(password.encode('utf-8'), salt=bcrypt.gensalt())

class User(Document):
    username = StringField(help_text="The username for the user", min_length=3, max_length=50)
    email = StringField(help_text="The email for the user", regex=re.compile(r'^\S+@\S+\.\S+$'))  # Simple email regex
    password = BinaryField(help_text="The hashed password for the user", hash_function=hash_password)
    age = IntegerField(help_text="The age of the user")
    status = EnumField(enum=Status, help_text="Indicator for whether the user is active or not.")

    class Meta:
        collection = "users"  # < If not provided, will default to class name (ex. User->user, UserDetails->user_details)
        created_at_timestamp = True  # < Provide a DateTimeField for document creation
        updated_at_timestamp = True  # < Provide a DateTimeField for document updates

Step 3: Create a MongoDB document using the User class

import bcrypt

await User.insert_one(
    {
        "username": "johndoe",
        "email": "[e-mail]",
        "password": "password123" #< hash_functon will hash the string literal password
    }
)

Step 4: Validate user was created in your MongoDB collection

You can do this using MongoDB compass, or alternatively, add a query to find all documents in the user collection after doing the insert in step 3:

users = User.find_many({})
if users:
    print("User collection contains the following documents:")
    for user in users:
        print(user.to_dict()) 
else:
    print("User collection failed to update! Check your MongoDB connection details and try again!")

Step 5: Put all the code above into one file and run it

python main.py

or in a FastAPI application:

uvicorn main:app --reload

Please refer to FastAPI Documentation for more details on how to get setup with FastAPI.

Congratulations 🎉

You've successfully created your first motormongo Object Document Mapper class. 🥳

The subsequent sections detail the datatype fields provided by motormongo, as well as the CRUD operations available on the classmethods and object instance methods of a motormongo document.

If you wish to get straight into how to integrate motormongo with your FastAPI application, skip ahead to the FastAPI Integration section.

motormongo Fields

motormongo supports the following datatype fields for your motormongo Document class:

1. StringField(min_length, max_length, regex)
2. IntegerField(min_value, max_value)
3. BooleanField()
4. EnumField(enum)
5. DateTimeField(auto_now, auto_now_add)
6. ListField()
7. EmbeddedDocumentField(EmbeddedDocument)
8. ReferenceField(Document)
9. BinaryField(hash_function: function)
10. GeoJSONField(return_as_json: bool)

Class methods

Operations

The following class methods are supported by motormongo's Document class:

CRUD Type	Operation
Create	insert_one(document: dict, **kwargs) -> Document
Create	insert_many(documents: List[dict]) -> Tuple[List[Document], Any]
Read	find_one(query: dict, **kwargs) -> Document
Read	find_many(query: dict, limit: int, **kwargs) -> List[Document]
Update	update_one(query: dict, update_fields: dict) -> Document
Update	update_many(query: dict, update_fields: dict) -> Tuple[List[Document], int]
Delete	delete_one(query: dict, **kwargs) -> bool
Delete	delete_many(query: dict, **kwargs) -> int
Mixed	find_one_or_create(query: dict, defaults: dict) -> Tuple[Document, bool]
Mixed	find_one_and_replace(query: dict, replacement: dict) -> Document
Mixed	find_one_and_delete(query: dict) -> Document
Mixed	find_one_and_update_empty_fields(query: dict, update_fields: dict) -> Tuple[Document, bool]

All examples below assume User is a subclass of motormongo provided Document class.

Create

insert_one(document: dict, **kwargs) -> Document

Inserts a single document into the database.

user = await User.insert_one({
    "name": "John",
    "age": 24,
    "alive": True
})

Alternatively, using **kwargs:

user = await User.insert_one(
 name="John",
 age=24,
 alive=True)

And similarly, with a dictionary:

user_document = {
    "name": "John",
    "age": 24,
    "alive": True
}
user = await User.insert_one(**user_document)

insert_many(List[document]) -> tuple[List['Document'], Any]

users, user_ids = await User.insert_many(
    [
        {
            "name": "John",
            "age": 24,
            "alive": True
        },
        {
            "name": "Mary",
            "age": 2,
            "alive": False
        }
    ]
)

or

docs_to_insert = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
inserted_docs, inserted_ids = await User.insert_many(docs_to_insert)

Read

find_one(query, **kwargs) -> Document

user = await User.find_one(
    {
        "_id": "655fc281c440f677fa1e117e"
    }
)

Alternatively, using **kwargs:

user = await User.find_one(_id="655fc281c440f677fa1e117e")

Note: The _id string datatype here is automatically converted to a BSON ObjectID, however, motormongo handles the scenario when a BSON ObjectId is passed as the _id datatype:

from bson import ObjectId

user = await User.find_one(
    {
        "_id": ObjectId("655fc281c440f677fa1e117e")
    }
)

find_many(query, limit, **kwargs) -> List[Document]

users =  await User.find_many(age={"$gt": 40}, alive=False, limit=20)

or

query = {"age": {"$gt": 40}, "alive": False}
users = await User.find_many(**query, limit=20)

Update

update_one(query, updated_fields) -> Document

updated_user = await User.update_one(
    {
        "_id": "655fc281c440f677fa1e117e"
    },
    {
        "name": "new_name",
        "age": 30
    }
)

or

query_criteria = {"name": "old_name"}
update_data = {"name": "updated_name"}
updated_user = await User.update_one(query_criteria, update_data)

update_many(qeury, fields) -> Tuple[List[Any], int]

updated_users, modified_count = await User.update_many({'age': {'$gt': 40}}, {'category': 'senior'})

another example:

updated_users, modified_count = await User.update_many({'name': 'John Doe'}, {'$inc': {'age': 1}})

Destroy

delete_one(query, **kwargs) -> bool

deleted = await User.delete_one({'_id': '507f191e810c19729de860ea'})

Alternatively, using **kwargs:

deleted = await User.delete_one(name='John Doe')

delete_many(query, **kwargs) -> int

deleted_count = await User.delete_many({'age': {'$gt': 40}})

Another example:

# Delete all users with a specific status
deleted_count = await User.delete_many({'status': 'inactive'})

Alternatively, using **kwargs:

deleted_count = await User.delete_many(status='inactive')

Mixed

find_one_or_create(query, defaults) -> Tuple['Document', bool]

user, created = await User.find_one_or_create({'username': 'johndoe'}, defaults={'age': 30})

find_one_and_replace(query, replacement) -> Document

replaced_user = await User.find_one_and_replace({'username': 'johndoe'}, {'username': 'johndoe', 'age': 35})

find_one_and_delete(query) -> Document

deleted_user = await User.find_one_and_delete({'username': 'johndoe'})

find_one_and_update_empty_fields(query, update_fields) -> Tuple['Document', bool]

updated_user, updated = await User.find_one_and_update_empty_fields(
                {'username': 'johndoe'},
                {'email': '[e-mail]', 'age': 30}
            )

Instance methods

motormongo also supports the manimulation of fields on the object instance. This allows users to programmatically achieve the same operations listed above through the object instance itself.

Operations

The following are object instance methods are supported by motormongo's Document class:

CRUD Type	Operation
Create	save() -> None
Delete	delete() -> None

NOTE: All update operations can be manipulated on the fields in the Document class object itself.

user.save() -> None

# Find user by MongoDB _id
user = await User.find_one(
    {
        "_id": "655fc281c440f677fa1e117e"
    }
)
# If there age is greater than 80, make them dead
if user.age > 80:
    user.alive = False
# Persist update on User instance in MongoDB mongo
user.save()

In this example, User.find_one() returns an instance of User. If the age field is greater than 80, the alive field is set to false. The instance of the document in the MongoDB database is then updated by calling the .save() method on the User object instance.

Destroy

user.delete() -> None

# Find all users where the user is not alive
users = await User.find_many(
    {
        "alive": False
    }
)
# Recursively delete all User instances in the users list who are not alive
for user in users:
    user.delete()

FastAPI integration

motormongo can be easily integrated in FastAPI APIs to leverage the asynchronous ability of FastAPI. To leverage motormongo's ease-of-use, Pydantic model's should be created to represent the MongoDB Document as a Pydantic model.

Below are some example APIs detailing how

Creating a document

from models.documents import User
from models.requests import UserModel


@app.post("/users/")
async def create_user(user: UserModel):
    new_user = User(**user.model_dump())
    await new_user.save()
    return new_user.to_dict()

Note:

License

This project is licensed under the MIT License.