Skip to main content

Asynchronous lightweight ODM for mongodb based on motor

Project description

monypy - asynchronous lightweight ODM for MongoDB based on motor

Build Status

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

monypy-2.0.tar.gz (4.9 kB view hashes)

Uploaded Source

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