Skip to content

Backend class reference

In most cases, you don't need to interact with backends directly, as they are used internally by Queue. For usual usage, see the Queue reference and Getting Started guides to learn how to configure and use different backends.

If you need to implement a custom backend or want to understand how existing backends work, here's the reference information for the backend classes.

sheppy.Backend

Bases: ABC

is_connected 

is_connected: bool

connect 

connect() -> None
Source code in src/sheppy/backend/base.py 
12
13
14
@abstractmethod
async def connect(self) -> None:
    pass

disconnect 

disconnect() -> None
Source code in src/sheppy/backend/base.py 
16
17
18
@abstractmethod
async def disconnect(self) -> None:
    pass

append 

append(
    queue_name: str,
    tasks: list[dict[str, Any]],
    unique: bool = True,
) -> list[bool]
Source code in src/sheppy/backend/base.py 
25
26
27
@abstractmethod
async def append(self, queue_name: str, tasks: list[dict[str, Any]], unique: bool = True) -> list[bool]:
    pass

pop 

pop(
    queue_name: str,
    limit: int = 1,
    timeout: float | None = None,
) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
29
30
31
@abstractmethod
async def pop(self, queue_name: str, limit: int = 1, timeout: float | None = None) -> list[dict[str, Any]]:
    pass

get_tasks 

get_tasks(
    queue_name: str, task_ids: list[str]
) -> dict[str, dict[str, Any]]
Source code in src/sheppy/backend/base.py 
33
34
35
@abstractmethod
async def get_tasks(self, queue_name: str, task_ids: list[str]) -> dict[str, dict[str, Any]]:
    pass

get_all_tasks 

get_all_tasks(queue_name: str) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
37
38
39
@abstractmethod
async def get_all_tasks(self, queue_name: str) -> list[dict[str, Any]]:
    pass

get_pending 

get_pending(
    queue_name: str, count: int = 1
) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
41
42
43
@abstractmethod
async def get_pending(self, queue_name: str, count: int = 1) -> list[dict[str, Any]]:
    pass

schedule 

schedule(
    queue_name: str,
    task_data: dict[str, Any],
    at: datetime,
    unique: bool = True,
) -> bool
Source code in src/sheppy/backend/base.py 
45
46
47
@abstractmethod
async def schedule(self, queue_name: str, task_data: dict[str, Any], at: datetime, unique: bool = True) -> bool:
    pass

get_scheduled 

get_scheduled(queue_name: str) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
49
50
51
@abstractmethod
async def get_scheduled(self, queue_name: str) -> list[dict[str, Any]]:
    pass

pop_scheduled 

pop_scheduled(
    queue_name: str, now: datetime | None = None
) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
53
54
55
@abstractmethod
async def pop_scheduled(self, queue_name: str, now: datetime | None = None) -> list[dict[str, Any]]:
    pass

store_result 

store_result(
    queue_name: str, task_data: dict[str, Any]
) -> bool
Source code in src/sheppy/backend/base.py 
57
58
59
@abstractmethod
async def store_result(self, queue_name: str, task_data: dict[str, Any]) -> bool:
    pass

get_results 

get_results(
    queue_name: str,
    task_ids: list[str],
    timeout: float | None = None,
) -> dict[str, dict[str, Any]]
Source code in src/sheppy/backend/base.py 
61
62
63
@abstractmethod
async def get_results(self, queue_name: str, task_ids: list[str], timeout: float | None = None) -> dict[str, dict[str, Any]]:
    pass

size 

size(queue_name: str) -> int
Source code in src/sheppy/backend/base.py 
65
66
67
@abstractmethod
async def size(self, queue_name: str) -> int:
    pass

clear 

clear(queue_name: str) -> int
Source code in src/sheppy/backend/base.py 
69
70
71
@abstractmethod
async def clear(self, queue_name: str) -> int:
    pass

get_stats 

get_stats(queue_name: str) -> dict[str, int]
Source code in src/sheppy/backend/base.py 
73
74
75
@abstractmethod
async def get_stats(self, queue_name: str) -> dict[str, int]:
    pass

list_queues 

list_queues() -> dict[str, int]
Source code in src/sheppy/backend/base.py 
77
78
79
@abstractmethod
async def list_queues(self) -> dict[str, int]:
    pass

add_cron 

add_cron(
    queue_name: str,
    deterministic_id: str,
    task_cron: dict[str, Any],
) -> bool
Source code in src/sheppy/backend/base.py 
81
82
83
@abstractmethod
async def add_cron(self, queue_name: str, deterministic_id: str, task_cron: dict[str, Any]) -> bool:
    pass

delete_cron 

delete_cron(queue_name: str, deterministic_id: str) -> bool
Source code in src/sheppy/backend/base.py 
85
86
87
@abstractmethod
async def delete_cron(self, queue_name: str, deterministic_id: str) -> bool:
    pass

get_crons 

get_crons(queue_name: str) -> list[dict[str, Any]]
Source code in src/sheppy/backend/base.py 
89
90
91
@abstractmethod
async def get_crons(self, queue_name: str) -> list[dict[str, Any]]:
    pass