monypy 1.0.3

Asynchronous lightweight ODM for mongodb

monypy - asynchronous lightweight ODM for MongoDB

Dependencies

python >= 3.6
motor >= 1.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.get_event_loop() \
    .run_until_complete(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.

• __loop__

  optional. Special event loop instance to use instead of default.

• __init_data__

  optional. Sets the initializing data for all objects in the collection when the object is initialized. If the value is callable, an instance will be passed to the value and called.

• manager

  The class attribute for database queries. Example:

  users_count = await User.manager.count()
  assert users_count == 1
  

• manager_class

  optional. Set a custom manager class. Learn more.

• _as_dict()

  Returns the object as a dict.

• save()

  сoroutine. Saves the object in the database if it does not exist, if it exists, it updates.

• delete()

  сoroutine. Removes an object from the database. If the object does not exist, then the DocumentDoesNotExistError exception is raised.

• refresh()

  сoroutine. Updates an object from the database. If the object does not exist, then the DocumentDoesNotExistError exception is raised.

Manager

A simple wrapper over AsyncIOMotorCollection. Example:

from monypy import Doc, Manager

class UserManager(Manager):
    def get_active(self):
        return self.find({'active': True})
        
        
class User(Doc):
    manager_class = UserManager

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

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

assert await User.manager.count() == 2
assert await User.manager.get_active().count() == 1

assert len([u async for u in User.manager.get_active()]) == 1