Redis Backend

class arend.backends.redis.RedisSettings(*, redis_host: str, redis_port: int = 6379, redis_db: int = 1, redis_password: str, redis_extras: Dict = None)

Defines settings for Redis Backend

get_backend() Type[RedisTask]

Returns a RedisTask class with settings already set

class arend.backends.redis.RedisTask(*, uuid: UUID = None, name: str, description: str = None, location: str = 'tasks', status: str = 'SCHEDULED', result: str = None, detail: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, args: tuple = None, kwargs: dict = None, queue: str, delay: int = 0, priority: int = 1, count_retries: int = 0, max_retries: int = 3, retry_backoff_factor: int = 1, created: datetime = None, updated: datetime = None)

RedisLog class. Defines the Log for Redis Backend

Usage:

>>> from arend.backends.redis import RedisSettings
>>>
>>> settings = RedisSettings(
>>>     redis_host="redis",
>>>     redis_port=6379,
>>>     redis_db="logs",
>>>     redis_password="pass"
>>> )
>>> RedisTask = RedisSettings.get_backend()  # type: Type[RedisLog]
>>> task = RedisTask(name="My task", description="A cool task")
>>> task.save()
>>>
>>> assert task.dict() == {"name": "My task", ...}
>>> assert task.json() == '{"name": "My task", ...}'
>>>
>>> task = RedisTask.get(uuid=UUID("<your-uuid>"))
>>> assert task.description == "A cool task"
>>>
>>> assert task.delete() == 1
delete() int

Deletes object in DataBase

Usage:

>>> assert task.delete() == 1  # count deleted 1
>>> assert task.delete() == 0  # count deleted 0
classmethod get(uuid: UUID) Optional[RedisTask]

Get object from DataBase

Usage:

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

Updates/Creates object in DataBase

Usage:

>>> task = RedisTask(name="My Task")
>>> task.save()
>>> task.description = "A new description"
>>> task.save()
class arend.backends.redis.RedisTasks(*, tasks: List[RedisTask] = None, count: int = 0)

Defines the RedisTasks collection