/
senorodion
/
Piway
Обзор
Документация
Войти
/
senorodion
/
Piway
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
develop
piway/workers/worker.py
48 строк
1 KB
Rodion Glukhov
Упрощение архитектуры
05 июл 2026, 16:38
Верифицирован
05 июл 2026, 16:38
57daa8a
Код
Авторство
О чём код?
from datetime import timedelta from threading import Event, Lock, Thread from typing import TYPE_CHECKING if TYPE_CHECKING: from piway.types_ import TaskType class Worker: interval = timedelta(seconds=30) def __init__(self, *, tasks: list[TaskType]) -> None: self._tasks = tasks self._thread = Thread(target=self._execute_tasks, name="worker") self._stop_event = Event() self._tasks_lock = Lock() def start(self) -> None: self._thread.start() def add_task(self, task: TaskType) -> None: with self._tasks_lock: self._tasks.append(task) def remove_task(self, task: TaskType) -> None: with self._tasks_lock: self._tasks.remove(task) def stop(self, *, wait_timeout: timedelta) -> None: self._stop_event.set() self._thread.join(timeout=wait_timeout.total_seconds()) if self._thread.is_alive(): raise TimeoutError def _execute_tasks(self) -> None: while not self._stop_event.is_set(): with self._tasks_lock: tasks = self._tasks.copy() for task in tasks: try: task() except Exception: pass if self._stop_event.wait(timeout=self.interval.total_seconds()): break