monypy

monypy - asynchronous lightweight ODM for MongoDB based on motor

Dependencies

python <= 3.7
motor >= 2.0

Installation

pipenv install monypy

Quick Start

import asyncio
from monypy import Doc


class User(Doc):
    __init_data__ = {
        'sex': 'male',
        'instance_id': lambda i: id(i)
    }
    
    __database__ = {
        'name': 'test',
        'host': 'localhost',
        'port': 27017
    }
    
user = User(name='John')

assert '_id' not in user
assert user.name == 'John'
assert user.sex == 'male'

assert not callable(user.instance_id)
assert user.instance_id == id(user)

asyncio.run(user.save())

assert '_id' in user

API Reference

Doc

• __database__

  Attribute for setting up the database. Parameters:

  • name - the name of the database

  List of other optional parameters here.

• __collection__

  optional. Attribute for setting up the collection. Parameters:

  • name - the name of the collection

• __abstract__

  optional. If True, then the collection will not create a connection to the database.

• __init_data__

  optional. Set the initializing data for all objects in the collection when the object is initialized. If the value is callable, an instance will be passed as an argument.

• save()

  сoroutine. It saves the object in the database.

• delete()

  сoroutine. It remove an object from the database. If the object does not exist in the database, then the DocumentDoesNotExist exception will be raised.

• refresh()

  сoroutine. Refresh the current object from the same object from the database. If the object does not exist in the database, then the DocumentDoesNotExist exception will be raised.

Manager

A simple wrapper over AsyncIOMotorCollection.

• create(**kwargs)

  сoroutine. Create an object and return it.

• count(filter, session=None, **kwargs)

  сoroutine. Simple alias on count_documents.

For example:

from monypy import Doc, Manager

class UserManager(Manager):
    async def count_active(self):
        return await self.count_documents({'active': True})

class SecondUserManager(Manager):
    async def count_not_active(self):
        return await self.count_documents({'active': False})
        
        
class User(Doc):
    documents = UserManager()
    second_documents = SecondUserManager()

    __database__ = {
        'name': 'test'
    }
    
    __init_data__ = {
        'active': True,
    }

await User().save()
await User(active=False).save()

assert await User.documents.count() == 2
assert await User.documents.count_active() == 1
assert await User.second_documents.count_not_active() == 1