SyncThreadWorker

Init module for sync worker.

SyncThreadWorker

Bases: BaseWorker, SyncPluginMixin

Воркер, Получающий из Брокера задачи и обрабатывающий их.

Пример

from qtasks import QueueTasks
from qtasks.workers import SyncThreadWorker

worker = SyncThreadWorker()
app = QueueTasks(worker=worker)

Source code in src/qtasks/workers/sync_worker.py

class SyncThreadWorker(BaseWorker, SyncPluginMixin):
    """
    Воркер, Получающий из Брокера задачи и обрабатывающий их.

    ## Пример

    ```python
    from qtasks import QueueTasks
    from qtasks.workers import SyncThreadWorker

    worker = SyncThreadWorker()
    app = QueueTasks(worker=worker)
    ```
    """

    def __init__(
        self,
        name: Annotated[
            str,
            Doc(
                """
                    Имя проекта. Это имя также используется воркером.

                    По умолчанию: `QueueTasks`.
                    """
            ),
        ] = "QueueTasks",
        broker: Annotated[
            Optional["BaseBroker"],
            Doc(
                """
                    Брокер.

                    По умолчанию: `qtasks.brokers.SyncRedisBroker`.
                    """
            ),
        ] = None,
        log: Annotated[
            Optional[Logger],
            Doc(
                """
                    Логгер.

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

                    По умолчанию: `qtasks.configs.config.QueueConfig`.
                    """
            ),
        ] = None,
        events: Annotated[
            Optional["BaseEvents"],
            Doc(
                """
                    События.

                    По умолчанию: `qtasks.events.SyncEvents`.
                    """
            ),
        ] = None,
    ):
        """Инициализация синхронного воркера.

        Args:
            name (str, optional): Имя проекта. По умолчанию: "QueueTasks".
            broker (BaseBroker, optional): Брокер. По умолчанию: `None`.
            log (Logger, optional): Логгер. По умолчанию: `None`.
            config (QueueConfig, optional): Конфиг. По умолчанию: `None`.
            events (BaseEvents, optional): События. По умолчанию: `qtasks.events.SyncEvents`.
        """
        super().__init__(name=name, broker=broker, log=log, config=config, events=events)
        self.events = self.events or SyncEvents()
        self.broker = broker or SyncRedisBroker(
            name=self.name, log=self.log, config=self.config
        )
        self.queue = PriorityQueue()
        self._tasks: Dict[str, TaskExecSchema] = {}
        self._stop_event = Event()
        self.lock = Lock()
        self.threads: List[Thread] = []
        self.semaphore = Semaphore(self.config.max_tasks_process)

        self.task_executor = SyncTaskExecutor

    def worker(
        self,
        number: Annotated[
            int,
            Doc(
                """
                    Номер Воркера.
                    """
            ),
        ],
    ) -> None:
        """Обработчик задач.

        Args:
            number (int): Номер Воркера.
        """
        self.events.fire("worker_running", worker=self, number=number)

        try:
            while not self._stop_event.is_set():
                with self.lock:
                    if self.queue.empty():
                        sleep(0.1)
                        continue
                    task_broker: TaskPrioritySchema = self.queue.get()

                if task_broker is None:
                    break
                Thread(
                    target=self._execute_task, args=(task_broker,), daemon=True
                ).start()

        finally:
            self.events.fire("worker_stopping", worker=self, number=number)

    def _execute_task(
        self,
        task_broker: Annotated[
            TaskPrioritySchema,
            Doc(
                """
                    Схема приоритетной задачи.
                    """
            ),
        ],
    ) -> None:
        """Выполняет задачу независимо.

        Args:
            task_broker (TaskPrioritySchema): Схема приоритетной задачи.
        """
        with self.semaphore:
            model = TaskStatusProcessSchema(
                task_name=task_broker.name,
                priority=task_broker.priority,
                created_at=task_broker.created_at,
                updated_at=time(),
                args=task_broker.args,
                kwargs=task_broker.kwargs,
            )

            task_func = self._task_exists(task_broker=task_broker)
            if not task_func:
                self.queue.task_done()
                return

            new_model = self._plugin_trigger(
                "worker_execute_before",
                worker=self,
                task_broker=task_broker,
                task_func=task_func,
                model=model,
                return_last=True
            )
            if new_model:
                model = new_model.get("model", model)

            self.broker.update(
                name=f"{self.name}:{task_broker.uuid}", mapping=asdict(model)
            )

            self.events.fire("task_running", worker=self, task_func=task_func, task_broker=task_broker)
            model = self._run_task(task_func=task_func, task_broker=task_broker)
            self.events.fire("task_stopping", worker=self, task_func=task_func, task_broker=task_broker, model=model)

            self.remove_finished_task(task_func=task_func, task_broker=task_broker, model=model)

            self._plugin_trigger("worker_execute_after", task_func=task_func, task_broker=task_broker, model=model)

            self.queue.task_done()

    def add(
        self,
        name: Annotated[
            str,
            Doc(
                """
                    Имя задачи.
                    """
            ),
        ],
        uuid: Annotated[
            UUID,
            Doc(
                """
                    UUID задачи.
                    """
            ),
        ],
        priority: Annotated[
            int,
            Doc(
                """
                    Приоритет задачи.
                    """
            ),
        ],
        created_at: Annotated[
            float,
            Doc(
                """
                    Создание задачи в формате timestamp.
                    """
            ),
        ],
        args: Annotated[
            tuple,
            Doc(
                """
                    Аргументы задачи типа args.
                    """
            ),
        ],
        kwargs: Annotated[
            dict,
            Doc(
                """
                    Аргументы задачи типа kwargs.
                    """
            ),
        ],
    ) -> Task:
        """Добавление задачи в очередь.

        Args:
            name (str): Имя задачи.
            uuid (UUID): UUID задачи.
            priority (int): Приоритет задачи.
            created_at (float): Создание задачи в формате timestamp.
            args (tuple): Аргументы задачи типа args.
            kwargs (dict): Аргументы задачи типа kwargs.
        """
        new_data = self._plugin_trigger(
            "worker_add",
            worker=self,
            task_name=name,
            uuid=uuid,
            priority=priority,
            args=args,
            kw=kwargs,
            created_at=created_at,
            return_last=True
        )
        if new_data:
            name = new_data.get("name", name)
            uuid = new_data.get("uuid", uuid)
            priority = new_data.get("priority", priority)
            args = new_data.get("args", args)
            kwargs = new_data.get("kw", kwargs)
            created_at = new_data.get("created_at", created_at)

        model = TaskPrioritySchema(
            priority=priority,
            uuid=uuid,
            name=name,
            args=list(args),
            kwargs=kwargs,
            created_at=created_at,
            updated_at=created_at,
        )
        with self.lock:
            self.queue.put(model)

        return Task(
            status=TaskStatusEnum.NEW.value,
            task_name=name,
            uuid=uuid,
            priority=priority,
            args=args,
            kwargs=kwargs,
            created_at=created_at,
            updated_at=created_at,
        )

    def start(
        self,
        num_workers: Annotated[
            int,
            Doc(
                """
                    Количество воркеров.

                    По умолчанию: `4`.
                    """
            ),
        ] = 4,
    ) -> None:
        """Запускает несколько обработчиков задач.

        Args:
            num_workers (int, optional): Количество воркеров. По умолчанию: 4.
        """
        self.num_workers = num_workers
        self._plugin_trigger("worker_start", worker=self)

        for number in range(self.num_workers):
            thread = Thread(target=self.worker, args=(number,), daemon=True)
            thread.start()
            self.threads.append(thread)

    def stop(self):
        """Останавливает воркеры."""
        self._plugin_trigger("worker_stop", worker=self)
        self._stop_event.set()
        for thread in self.threads:
            thread.join()

    def update_config(self, config: QueueConfig):
        """Обновляет конфиг."""
        self.config = config
        self.semaphore = Semaphore(config.max_tasks_process)

    def _run_task(
        self, task_func: TaskExecSchema, task_broker: TaskPrioritySchema
    ) -> Union[TaskStatusSuccessSchema, TaskStatusErrorSchema, TaskStatusCancelSchema]:
        """Запуск функции задачи.

        Args:
            task_func (TaskExecSchema): Схема `qtasks.schemas.TaskExecSchema`.
            task_broker (TaskPrioritySchema): Схема `qtasks.schemas.TaskPrioritySchema`.

        Returns:
            TaskStatusSuccessSchema|TaskStatusErrorSchema: Результат функции задачи.
        """
        self.log.info(
            f"Выполняю задачу {task_broker.uuid} ({task_broker.name}), приоритет: {task_broker.priority}"
        )

        new_data = self._plugin_trigger(
            "worker_run_task_before",
            worker=self,
            task_func=task_func,
            task_broker=task_broker,
            return_last=True
        )
        if new_data:
            task_func = new_data.get("task_func", task_func)
            task_broker = new_data.get("task_broker", task_broker)

        if self.task_middlewares_before:
            task_func.add_middlewares_before(self.task_middlewares_before)
        if self.task_middlewares_after:
            task_func.add_middlewares_after(self.task_middlewares_after)

        executor = task_func.executor if task_func.executor is not None else self.task_executor
        executor = executor(
            task_func=task_func,
            task_broker=task_broker,
            log=self.log,
            plugins=self.plugins,
        )

        try:
            result = executor.execute()
            return self._task_success(result, task_func, task_broker)
        except TaskCancelError as e:
            return self._task_cancel(e, task_func, task_broker)
        except BaseException as e:
            return self._task_error(e, task_func, task_broker)

    def _task_success(self, result: Any, task_func: TaskExecSchema, task_broker: TaskPrioritySchema) -> None:
        """Событие успешного завершения задачи."""
        model = TaskStatusSuccessSchema(
            task_name=task_func.name,
            priority=task_func.priority,
            returning=result,
            created_at=task_broker.created_at,
            updated_at=time(),
            args=task_broker.args,
            kwargs=task_broker.kwargs,
        )
        self.log.info(
            f"Задача {task_broker.uuid} успешно завершена, результат: {result}"
        )
        return model

    def _task_error(self, e, task_func: TaskExecSchema, task_broker: TaskPrioritySchema) -> None:
        """Событие завершения задачи с ошибкой."""
        trace = traceback.format_exc()

        # plugin: retry
        plugin_result = None

        should_retry = (
            task_func.retry and (
                not task_func.retry_on_exc or type(e) in task_func.retry_on_exc
            )
        )
        if should_retry and task_func.retry:
                plugin_result = self._plugin_trigger(
                    "worker_task_error_retry",
                    broker=self.broker,
                    task_func=task_func,
                    task_broker=task_broker,
                    trace=trace,
                    return_last=True
                )

        if not plugin_result:
            model = TaskStatusErrorSchema(
                task_name=task_func.name,
                priority=task_func.priority,
                traceback=trace,
                created_at=task_broker.created_at,
                updated_at=time(),
                args=task_broker.args,
                kwargs=task_broker.kwargs,
            )
        else:
            model: TaskStatusErrorSchema = plugin_result.get("model", model)
        #

        self.log.error(f"Задача {task_broker.uuid} завершена с ошибкой:\n{trace}")
        return model

    def _task_cancel(self, e, task_func: TaskExecSchema, task_broker: TaskPrioritySchema) -> None:
        """Событие отмены задачи."""
        model = TaskStatusCancelSchema(
            task_name=task_func.name,
            priority=task_func.priority,
            cancel_reason=str(e),
            created_at=task_broker.created_at,
            updated_at=time(),
        )
        self.log.info(f"Задача {task_broker.uuid} была отменена по причине: {e}")
        return model

    def _task_exists(self, task_broker: TaskPrioritySchema) -> Union[TaskExecSchema, None]:
        """Проверка существования задачи.

        Args:
            task_broker (TaskPrioritySchema): Схема `TaskPrioritySchema`.

        Returns:
            TaskExecSchema|None: Схема `TaskExecSchema` или `None`.
        """
        try:
            return self._tasks[task_broker.name]
        except KeyError as e:
            self.log.warning(f"Задачи {e.args[0]} не существует!")
            trace = traceback.format_exc()
            model = TaskStatusErrorSchema(
                task_name=task_broker.name,
                priority=task_broker.priority,
                traceback=trace,
                created_at=task_broker.created_at,
                updated_at=time(),
            )
            self.remove_finished_task(task_func=None, task_broker=task_broker, model=model)
            self.log.error(f"Задача {task_broker.name} завершена с ошибкой:\n{trace}")
            return None

    def remove_finished_task(
        self,
        task_func: Annotated[
            Optional[TaskExecSchema],
            Doc(
                """
                    Схема функции задачи.
                    """
            )
        ],
        task_broker: Annotated[
            TaskPrioritySchema,
            Doc(
                """
                    Схема приоритетной задачи.
                    """
            ),
        ],
        model: Annotated[
            Union[
                TaskStatusProcessSchema, TaskStatusErrorSchema, TaskStatusCancelSchema
            ],
            Doc(
                """
                    Модель результата задачи.
                    """
            ),
        ],
    ) -> None:
        """Обновляет данные хранилища через функцию `self.storage.remove_finished_task`.

        Args:
            task_func (TaskExecSchema, optional): Схема функции задачи. По умолчанию: `None`.
            task_broker (TaskPrioritySchema): Схема приоритетной задачи.
            model (TaskStatusNewSchema | TaskStatusErrorSchema | TaskStatusCancelSchema): Модель результата задачи.
        """
        new_data = self._plugin_trigger(
            "worker_remove_finished_task",
            worker=self,
            broker=self.broker,
            task_func=task_func,
            task_broker=task_broker,
            model=model,
            return_last=True
        )
        if new_data:
            task_broker, model = new_data.get("task_broker", task_broker), new_data.get("model", model)
        self.broker.remove_finished_task(task_broker, model)

    def init_plugins(self):
        """Инициализация плагинов."""
        self.add_plugin(SyncRetryPlugin(), trigger_names=["worker_task_error_retry"])
        self.add_plugin(SyncPydanticWrapperPlugin(), trigger_names=["task_executor_args_replace", "task_executor_after_execute_result_replace"])
        self.add_plugin(SyncDependsPlugin(), trigger_names=["task_executor_args_replace"])

__init__(name='QueueTasks', broker=None, log=None, config=None, events=None)

Инициализация синхронного воркера.

Parameters:

Name	Type	Description	Default
name	str	Имя проекта. По умолчанию: "QueueTasks".	'QueueTasks'
broker	BaseBroker	Брокер. По умолчанию: None.	None
log	Logger	Логгер. По умолчанию: None.	None
config	QueueConfig	Конфиг. По умолчанию: None.	None
events	BaseEvents	События. По умолчанию: qtasks.events.SyncEvents.	None

Source code in src/qtasks/workers/sync_worker.py

def __init__(
    self,
    name: Annotated[
        str,
        Doc(
            """
                Имя проекта. Это имя также используется воркером.

                По умолчанию: `QueueTasks`.
                """
        ),
    ] = "QueueTasks",
    broker: Annotated[
        Optional["BaseBroker"],
        Doc(
            """
                Брокер.

                По умолчанию: `qtasks.brokers.SyncRedisBroker`.
                """
        ),
    ] = None,
    log: Annotated[
        Optional[Logger],
        Doc(
            """
                Логгер.

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

                По умолчанию: `qtasks.configs.config.QueueConfig`.
                """
        ),
    ] = None,
    events: Annotated[
        Optional["BaseEvents"],
        Doc(
            """
                События.

                По умолчанию: `qtasks.events.SyncEvents`.
                """
        ),
    ] = None,
):
    """Инициализация синхронного воркера.

    Args:
        name (str, optional): Имя проекта. По умолчанию: "QueueTasks".
        broker (BaseBroker, optional): Брокер. По умолчанию: `None`.
        log (Logger, optional): Логгер. По умолчанию: `None`.
        config (QueueConfig, optional): Конфиг. По умолчанию: `None`.
        events (BaseEvents, optional): События. По умолчанию: `qtasks.events.SyncEvents`.
    """
    super().__init__(name=name, broker=broker, log=log, config=config, events=events)
    self.events = self.events or SyncEvents()
    self.broker = broker or SyncRedisBroker(
        name=self.name, log=self.log, config=self.config
    )
    self.queue = PriorityQueue()
    self._tasks: Dict[str, TaskExecSchema] = {}
    self._stop_event = Event()
    self.lock = Lock()
    self.threads: List[Thread] = []
    self.semaphore = Semaphore(self.config.max_tasks_process)

    self.task_executor = SyncTaskExecutor

add(name, uuid, priority, created_at, args, kwargs)

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

Parameters:

Name	Type	Description	Default
name	str	Имя задачи.	required
uuid	UUID	UUID задачи.	required
priority	int	Приоритет задачи.	required
created_at	float	Создание задачи в формате timestamp.	required
args	tuple	Аргументы задачи типа args.	required
kwargs	dict	Аргументы задачи типа kwargs.	required

Source code in src/qtasks/workers/sync_worker.py

def add(
    self,
    name: Annotated[
        str,
        Doc(
            """
                Имя задачи.
                """
        ),
    ],
    uuid: Annotated[
        UUID,
        Doc(
            """
                UUID задачи.
                """
        ),
    ],
    priority: Annotated[
        int,
        Doc(
            """
                Приоритет задачи.
                """
        ),
    ],
    created_at: Annotated[
        float,
        Doc(
            """
                Создание задачи в формате timestamp.
                """
        ),
    ],
    args: Annotated[
        tuple,
        Doc(
            """
                Аргументы задачи типа args.
                """
        ),
    ],
    kwargs: Annotated[
        dict,
        Doc(
            """
                Аргументы задачи типа kwargs.
                """
        ),
    ],
) -> Task:
    """Добавление задачи в очередь.

    Args:
        name (str): Имя задачи.
        uuid (UUID): UUID задачи.
        priority (int): Приоритет задачи.
        created_at (float): Создание задачи в формате timestamp.
        args (tuple): Аргументы задачи типа args.
        kwargs (dict): Аргументы задачи типа kwargs.
    """
    new_data = self._plugin_trigger(
        "worker_add",
        worker=self,
        task_name=name,
        uuid=uuid,
        priority=priority,
        args=args,
        kw=kwargs,
        created_at=created_at,
        return_last=True
    )
    if new_data:
        name = new_data.get("name", name)
        uuid = new_data.get("uuid", uuid)
        priority = new_data.get("priority", priority)
        args = new_data.get("args", args)
        kwargs = new_data.get("kw", kwargs)
        created_at = new_data.get("created_at", created_at)

    model = TaskPrioritySchema(
        priority=priority,
        uuid=uuid,
        name=name,
        args=list(args),
        kwargs=kwargs,
        created_at=created_at,
        updated_at=created_at,
    )
    with self.lock:
        self.queue.put(model)

    return Task(
        status=TaskStatusEnum.NEW.value,
        task_name=name,
        uuid=uuid,
        priority=priority,
        args=args,
        kwargs=kwargs,
        created_at=created_at,
        updated_at=created_at,
    )

init_plugins()

Инициализация плагинов.

Source code in src/qtasks/workers/sync_worker.py

def init_plugins(self):
    """Инициализация плагинов."""
    self.add_plugin(SyncRetryPlugin(), trigger_names=["worker_task_error_retry"])
    self.add_plugin(SyncPydanticWrapperPlugin(), trigger_names=["task_executor_args_replace", "task_executor_after_execute_result_replace"])
    self.add_plugin(SyncDependsPlugin(), trigger_names=["task_executor_args_replace"])

remove_finished_task(task_func, task_broker, model)

Обновляет данные хранилища через функцию self.storage.remove_finished_task.

Parameters:

Name	Type	Description	Default
task_func	TaskExecSchema	Схема функции задачи. По умолчанию: None.	required
task_broker	TaskPrioritySchema	Схема приоритетной задачи.	required
model	TaskStatusNewSchema | TaskStatusErrorSchema | TaskStatusCancelSchema	Модель результата задачи.	required

Source code in src/qtasks/workers/sync_worker.py

def remove_finished_task(
    self,
    task_func: Annotated[
        Optional[TaskExecSchema],
        Doc(
            """
                Схема функции задачи.
                """
        )
    ],
    task_broker: Annotated[
        TaskPrioritySchema,
        Doc(
            """
                Схема приоритетной задачи.
                """
        ),
    ],
    model: Annotated[
        Union[
            TaskStatusProcessSchema, TaskStatusErrorSchema, TaskStatusCancelSchema
        ],
        Doc(
            """
                Модель результата задачи.
                """
        ),
    ],
) -> None:
    """Обновляет данные хранилища через функцию `self.storage.remove_finished_task`.

    Args:
        task_func (TaskExecSchema, optional): Схема функции задачи. По умолчанию: `None`.
        task_broker (TaskPrioritySchema): Схема приоритетной задачи.
        model (TaskStatusNewSchema | TaskStatusErrorSchema | TaskStatusCancelSchema): Модель результата задачи.
    """
    new_data = self._plugin_trigger(
        "worker_remove_finished_task",
        worker=self,
        broker=self.broker,
        task_func=task_func,
        task_broker=task_broker,
        model=model,
        return_last=True
    )
    if new_data:
        task_broker, model = new_data.get("task_broker", task_broker), new_data.get("model", model)
    self.broker.remove_finished_task(task_broker, model)

start(num_workers=4)

Запускает несколько обработчиков задач.

Parameters:

Name	Type	Description	Default
num_workers	int	Количество воркеров. По умолчанию: 4.	4

Source code in src/qtasks/workers/sync_worker.py

def start(
    self,
    num_workers: Annotated[
        int,
        Doc(
            """
                Количество воркеров.

                По умолчанию: `4`.
                """
        ),
    ] = 4,
) -> None:
    """Запускает несколько обработчиков задач.

    Args:
        num_workers (int, optional): Количество воркеров. По умолчанию: 4.
    """
    self.num_workers = num_workers
    self._plugin_trigger("worker_start", worker=self)

    for number in range(self.num_workers):
        thread = Thread(target=self.worker, args=(number,), daemon=True)
        thread.start()
        self.threads.append(thread)

stop()

Останавливает воркеры.

Source code in src/qtasks/workers/sync_worker.py

def stop(self):
    """Останавливает воркеры."""
    self._plugin_trigger("worker_stop", worker=self)
    self._stop_event.set()
    for thread in self.threads:
        thread.join()

update_config(config)

Обновляет конфиг.

Source code in src/qtasks/workers/sync_worker.py

def update_config(self, config: QueueConfig):
    """Обновляет конфиг."""
    self.config = config
    self.semaphore = Semaphore(config.max_tasks_process)

worker(number)

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

Parameters:

Name	Type	Description	Default
number	int	Номер Воркера.	required

Source code in src/qtasks/workers/sync_worker.py

def worker(
    self,
    number: Annotated[
        int,
        Doc(
            """
                Номер Воркера.
                """
        ),
    ],
) -> None:
    """Обработчик задач.

    Args:
        number (int): Номер Воркера.
    """
    self.events.fire("worker_running", worker=self, number=number)

    try:
        while not self._stop_event.is_set():
            with self.lock:
                if self.queue.empty():
                    sleep(0.1)
                    continue
                task_broker: TaskPrioritySchema = self.queue.get()

            if task_broker is None:
                break
            Thread(
                target=self._execute_task, args=(task_broker,), daemon=True
            ).start()

    finally:
        self.events.fire("worker_stopping", worker=self, number=number)