Features:
- Asynchronous
All example run into:
python -m asynciofrom ampo import CollectionWorker, AMPODatabase, ORMConfig, init_collection
# Initilize DB before calls db methods
AMPODatabase(url="mongodb://test")
# Pydantic Model
class ModelA(CollectionWorker):
field1: str
field2: int
model_config = ORMConfig(
orm_collection="test"
)
await init_collection()
inst_a = ModelA("test", 123)
await inst_a.save()
# Get object
inst_a = await ModelA.get(field1="test")For search by 'id' usages in filter '_id' or 'id' name.
# import
# Initilize DB before calls db methods
AMPODatabase(url="mongodb://test")
# Pydantic Model
class ModelA(CollectionWorker):
field1: str
model_config = ORMConfig(
orm_collection="test",
orm_indexes=[
{
"keys": ["field1"],
"options": {
"unique": True
}
}
]
)
# This method create indexes
# Call only one time
await init_collection()Suppport options:
- unique
- expireAfterSeconds
Keys is list of fields.
It works only with single field (TTL Indexes).
You should set the option 'expireAfterSeconds', and field 'keys' should have only single field.
Example:
# import
# Initilize DB before calls db methods
AMPODatabase(url="mongodb://test")
# Pydantic Model
class ModelA(CollectionWorker):
field1: datetime
model_config = ORMConfig(
orm_collection="test",
orm_indexes=[
{
"keys": ["field1"],
"options": {
"expireAfterSeconds": 10
}
}
]
)
# optional, set new value
ModelA.update_expiration_value("field1", 20)
await init_collection()if you want to set the 'expireAfterSeconds' only from method 'update_expiration_value', set it to '-1'. if you want skip the index changed, call method 'expiration_index_skip' before init_collection.
It is supported by default. Just, you need create the embedded document as class of pydantic - 'BaseModel'. It will be stored into db as object.
Example:
from pydantic import BaseModel
class Embeded(BaseModel):
name: str
class ModelA(CollectionWorker):
field1: str
field2: Embeded
model_config = ORMConfig(
orm_collection="test"
)Style:
Run tests:
env TEST_MONGO_URL=mongodb://localhost/test pytest
