Asynchronous lightweight ODM for mongodb based on motor
Project description
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
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.