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
- 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