Перейти к содержанию

BaseTimer - Пишем свой Таймер

Base timer class.

BaseTimer

Bases: ABC

BaseTimer - Абстрактный класс, который является фундаментом для Таймеров.

Пример

from qtasks import QueueTasks
from qtasks.timers.base import BaseTimer

class MyTimer(BaseTimer):
    def __init__(self, app: QueueTasks):
        super().__init__(app=app)
        pass
Source code in src/qtasks/timers/base.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class BaseTimer(ABC):
    """
    `BaseTimer` - Абстрактный класс, который является фундаментом для Таймеров.

    ## Пример

    ```python
    from qtasks import QueueTasks
    from qtasks.timers.base import BaseTimer

    class MyTimer(BaseTimer):
        def __init__(self, app: QueueTasks):
            super().__init__(app=app)
            pass
    ```
    """

    def __init__(
        self,
        app: Annotated[
            "QueueTasks",
            Doc(
                """
                    Задача.

                    По умолчанию: `{qtasks.QueueTasks}` или `{qtasks.asyncio.QueueTasks}`.
                    """
            ),
        ],
        log: Annotated[
            Optional[Logger],
            Doc(
                """
                    Логгер.

                    По умолчанию: `qtasks.logs.Logger`.
                    """
            ),
        ] = None,
        config: Annotated[
            Optional[QueueConfig],
            Doc(
                """
                    Конфиг.

                    По умолчанию: `qtasks.configs.config.QueueConfig`.
                    """
            ),
        ] = None,
    ):
        """Инициализация таймера.

        Args:
            app (QueueTasks): Приложение.
            log (Logger, optional): Логгер. По умолчанию: `qtasks.logs.Logger`.
            config (QueueConfig, optional): Конфиг. По умолчанию: `qtasks.configs.config.QueueConfig`.
        """
        self.app = app
        self.config = config or self.app.config
        self.log = (
            log.with_subname("Timer")
            if log
            else Logger(
                name=self.app.name,
                subname="Timer",
                default_level=self.config.logs_default_level_server,
                format=self.config.logs_format,
            )
        )
        self.scheduler = AsyncIOScheduler()

    @abstractmethod
    def add_task(
        self,
        *args: Annotated[
            Optional[tuple],
            Doc(
                """
                    args задачи.

                    По умолчанию: `()`.
                    """
            ),
        ],
        task_name: Annotated[
            str,
            Doc(
                """
                    Имя задачи.
                    """
            ),
        ],
        priority: Annotated[
            Optional[int],
            Doc(
                """
                    Приоритет у задачи.

                    По умолчанию: Значение приоритета у задачи.
                    """
            ),
        ] = None,
        timeout: Annotated[
            Optional[float],
            Doc(
                """
                    Таймаут задачи.

                    Если указан, задача возвращается через `qtasks.results.AsyncTask`.
                    """
            ),
        ] = None,
        trigger: Annotated[
            Any,
            Doc(
                """
                    Триггер задачи.
                    """
            ),
        ],
        **kwargs: Annotated[
            Optional[dict],
            Doc(
                """
                    kwargs задачи.

                    По умолчанию: `{}`.
                    """
            ),
        ],
    ) -> Union[Any, None]:
        """Добавление задачи.

        Args:
            task_name (str): Имя задачи.
            trigger (Any, optional): Значения триггера.
            priority (int, optional): Приоритет задачи. По умолчанию `0`.
            args (tuple, optional): args задачи. По умолчанию `()`.
            kwargs (dict, optional): kwags задачи. По умолчанию `{}`.

        Returns:
            Any|None: Задача.
        """
        pass

    def run_forever(self):
        """Запуск Таймера."""
        pass

__init__(app, log=None, config=None)

Инициализация таймера.

Parameters:

Name Type Description Default
app QueueTasks

Приложение.

 required
log Logger

Логгер. По умолчанию: qtasks.logs.Logger.

 None
config QueueConfig

Конфиг. По умолчанию: qtasks.configs.config.QueueConfig.

 None
Source code in src/qtasks/timers/base.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    app: Annotated[
        "QueueTasks",
        Doc(
            """
                Задача.

                По умолчанию: `{qtasks.QueueTasks}` или `{qtasks.asyncio.QueueTasks}`.
                """
        ),
    ],
    log: Annotated[
        Optional[Logger],
        Doc(
            """
                Логгер.

                По умолчанию: `qtasks.logs.Logger`.
                """
        ),
    ] = None,
    config: Annotated[
        Optional[QueueConfig],
        Doc(
            """
                Конфиг.

                По умолчанию: `qtasks.configs.config.QueueConfig`.
                """
        ),
    ] = None,
):
    """Инициализация таймера.

    Args:
        app (QueueTasks): Приложение.
        log (Logger, optional): Логгер. По умолчанию: `qtasks.logs.Logger`.
        config (QueueConfig, optional): Конфиг. По умолчанию: `qtasks.configs.config.QueueConfig`.
    """
    self.app = app
    self.config = config or self.app.config
    self.log = (
        log.with_subname("Timer")
        if log
        else Logger(
            name=self.app.name,
            subname="Timer",
            default_level=self.config.logs_default_level_server,
            format=self.config.logs_format,
        )
    )
    self.scheduler = AsyncIOScheduler()

add_task(*args, task_name, priority=None, timeout=None, trigger, **kwargs) abstractmethod

Добавление задачи.

Parameters:

Name Type Description Default
task_name str

Имя задачи.

 required
trigger Any

Значения триггера.

 required
priority int

Приоритет задачи. По умолчанию 0.

 None
args tuple

args задачи. По умолчанию ().

 ()
kwargs dict

kwags задачи. По умолчанию {}.

 {}

Returns:

Type Description
Union[Any, None]

Any|None: Задача.
Source code in src/qtasks/timers/base.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@abstractmethod
def add_task(
    self,
    *args: Annotated[
        Optional[tuple],
        Doc(
            """
                args задачи.

                По умолчанию: `()`.
                """
        ),
    ],
    task_name: Annotated[
        str,
        Doc(
            """
                Имя задачи.
                """
        ),
    ],
    priority: Annotated[
        Optional[int],
        Doc(
            """
                Приоритет у задачи.

                По умолчанию: Значение приоритета у задачи.
                """
        ),
    ] = None,
    timeout: Annotated[
        Optional[float],
        Doc(
            """
                Таймаут задачи.

                Если указан, задача возвращается через `qtasks.results.AsyncTask`.
                """
        ),
    ] = None,
    trigger: Annotated[
        Any,
        Doc(
            """
                Триггер задачи.
                """
        ),
    ],
    **kwargs: Annotated[
        Optional[dict],
        Doc(
            """
                kwargs задачи.

                По умолчанию: `{}`.
                """
        ),
    ],
) -> Union[Any, None]:
    """Добавление задачи.

    Args:
        task_name (str): Имя задачи.
        trigger (Any, optional): Значения триггера.
        priority (int, optional): Приоритет задачи. По умолчанию `0`.
        args (tuple, optional): args задачи. По умолчанию `()`.
        kwargs (dict, optional): kwags задачи. По умолчанию `{}`.

    Returns:
        Any|None: Задача.
    """
    pass

run_forever()

Запуск Таймера.

Source code in src/qtasks/timers/base.py 
160
161
162
def run_forever(self):
    """Запуск Таймера."""
    pass