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