BaseTaskExecutor - Пишем свой исполнитель задач

Base Task Executor.

BaseTaskExecutor

Bases: ABC

BaseTaskExecutor - Абстрактный класс, который является фундаментом для классов исполнителей задач.

Пример

from qtasks.executors.base import BaseTaskExecutor

class MyTaskExecutor(BaseTaskExecutor):
    def __init__(self, name: str):
        super().__init__(name=name)
        pass

__init__(task_func, task_broker, middlewares=None, log=None, plugins=None)

Инициализация класса. Происходит внутри Worker перед обработкой задачи.

Parameters:

Name	Type	Description	Default
task_func	TaskExecSchema	Схема TaskExecSchema.	required
task_broker	TaskPrioritySchema	Схема TaskPrioritySchema.	required
middlewares	List[Type[TaskMiddleware]]	description. По умолчанию None.	None
log	Logger	класс qtasks.logs.Logger. По умолчанию: qtasks._state.log_main.	None

add_plugin(plugin, trigger_names=None)

Добавить плагин в класс.

Parameters:

Name	Type	Description	Default
plugin	BasePlugin	Плагин	required
trigger_names	List[str]	Имя триггеров для плагина. По умолчанию: будет добавлен в Globals.	None

after_execute()

Вызывается после выполнения задачи.

before_execute()

Вызывается перед выполнением задачи.

decode()

Декодирование задачи.

Returns:

Name	Type	Description
str	str	Результат задачи.

execute(decode=True) abstractmethod

Обработка задачи.

Parameters:

Name	Type	Description	Default
decode	bool	Декодирование. По умолчанию: True.	True

Returns:

Type	Description
Any | str	Any|str: Результат задачи.

execute_middlewares()

Вызов мидлварей.

run_task()

Вызов задачи.

Returns:

Name	Type	Description
Any	Any	Результат задачи.