Mongo Backend

Mongo Settings

class progress_updater.backends.mongo.MongoSettings(*, mongo_connection: str, mongo_db: str, mongo_collection: str, mongo_extras: Dict = None)

Mongo Settings. Defines settings for Mongo Backend

backend() → Type[MongoLog]

Returns a MongoLog class and set Mongo backend settings

Usage:

>>> from progress_updater.backends import MongoSettings
>>>
>>> settings = MongoSettings(
>>>     mongo_connection="mongodb://user:pass@mongo:27017",
>>>     mongo_db="db",
>>>     mongo_collection="logs"
>>> )
>>> MongoLog = MongoSettings.backend()  # type: Type[MongoLog]
>>> log = MongoLog(task_name="My task", description="A cool task")
>>> log.save()

Mongo Log

class progress_updater.backends.mongo.MongoLog(*, uuid: UUID = None, task_name: str, status: str = 'PENDING', log: str = '', description: str = None, start_time: datetime = None, end_time: datetime = None, created: datetime = None, updated: datetime = None)

MongoLog class. Defines the Log for Mongo Backend

Usage:

>>> from progress_updater.backends import MongoSettings
>>>
>>> settings = MongoSettings(
>>>     mongo_connection="mongodb://user:pass@mongo:27017",
>>>     mongo_db="db",
>>>     mongo_collection="logs"
>>> )
>>> MongoLog = MongoSettings.backend()  # type: Type[MongoLog]
>>> log = MongoLog(task_name="My task", description="A cool task")
>>> log.save()
>>>
>>> assert log.dict() == {"task_name": "My task", ...}
>>> assert log.json() == '{"task_name": "My task", ...}'
>>>
>>> log = MongoLog.get(uuid=UUID("<your-uuid>"))
>>> assert log.description == "A cool task"
>>>
>>> assert log.delete() == 1

delete() → int

Deletes object in DataBase

Usage:

>>> ...
>>> assert log.delete() == 1  # count deleted 1
>>> assert log.delete() == 0  # count deleted 0
>>> ...

classmethod get(uuid: UUID) → Optional[MongoLog]

Get object from DataBase

Usage:

>>> ...
>>> log = MongoLog.get(uuid=UUID("<your-uuid>"))
>>> assert log.uuid == UUID("<your-uuid>")
>>>

save() → MongoLog

Updates/Creates object in DataBase

Usage:

>>> ...
>>> log = MongoLog(task_name="My Task")
>>> log.save()
>>> log.description = "A new description"
>>> log.save()
>>> ...