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

QueueConfig

QueueConfig Schema.

QueueConfig dataclass

Конфигурация очередей задач.

Attributes:

Name Type Description
max_tasks_process int

Максимум задач в процессе. По умолчанию: 10
running_older_tasks bool

Запустить прошлые задачи. По умолчанию: False
delete_finished_tasks bool

Удаление выполненных задач. По умолчанию: False
default_task_priority int

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

Уровень логирования для сервера. По умолчанию: logging.INFO (20)
logs_default_level_client int | None

Уровень логирования для клиента. По умолчанию: logging.INFO (20)
logs_format str

Формат логирования. По умолчанию: "%(asctime)s [%(name)s: %(levelname)s] %(message)s"
Source code in src/qtasks/configs/config.py 
10
11
12
13
14
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
@dataclass
class QueueConfig:
    """
    Конфигурация очередей задач.

    Attributes:
        max_tasks_process (int): Максимум задач в процессе. По умолчанию: 10
        running_older_tasks (bool): Запустить прошлые задачи. По умолчанию: False
        delete_finished_tasks (bool): Удаление выполненных задач. По умолчанию: False

        default_task_priority (int): Приоритет задач по умолчанию. По умолчанию: 0

        logs_default_level_server (int): Уровень логирования для сервера. По умолчанию: logging.INFO (20)
        logs_default_level_client (int | None): Уровень логирования для клиента. По умолчанию: logging.INFO (20)
        logs_format (str): Формат логирования. По умолчанию: "%(asctime)s [%(name)s: %(levelname)s] %(message)s"
    """

    max_tasks_process: int = 10
    running_older_tasks: bool = False
    delete_finished_tasks: bool = True

    default_task_priority: int = 0

    global_config_status_ttl = 20
    global_config_status_set_periodic = 17

    logs_default_level_server: int = logging.INFO
    logs_default_level_client: int = logging.INFO
    logs_format: str = "%(asctime)s [%(name)s: %(levelname)s] (%(subname)s) %(message)s"

    result_time_interval: float = 1.0
    result_statuses_end: List[str] = field(
        default_factory=lambda: [
            TaskStatusEnum.SUCCESS.value,
            TaskStatusEnum.ERROR.value,
            TaskStatusEnum.CANCEL.value
        ]
    )

    _callbacks: List[Callable[["QueueConfig", str, Any], None]] = field(
        default_factory=list, init=False, repr=False
    )
    _dynamic_fields: Dict[str, Any] = field(
        default_factory=dict, init=False, repr=False
    )

    def subscribe(self, callback: Callable[["QueueConfig", str, Any], None]):
        """Подписка на изменение."""
        self._callbacks.append(callback)

    def _notify(self, key: str, value: Any):
        for callback in self._callbacks:
            callback(self, key, value)

    def __getattr__(self, item):
        """Получение атрибута."""
        # Проверяем динамические поля
        if item in self._dynamic_fields:
            return self._dynamic_fields[item]
        # Вызов стандартного поведения dataclass
        raise AttributeError(f"{type(self).__name__} has no attribute '{item}'")

    def __setattr__(self, key, value):
        """Установка атрибута."""
        # Для полей dataclass
        if key in self.__annotations__:
            object.__setattr__(self, key, value)
            if (
                "_callbacks" in self.__dict__
            ):  # Проверяем наличие для избежания рекурсии
                self._notify(key, value)
        # Для внутренних атрибутов
        elif key.startswith("_"):
            object.__setattr__(self, key, value)
        # Для динамических полей
        else:
            dynamic_fields = object.__getattribute__(self, "_dynamic_fields")
            dynamic_fields[key] = value
            self._notify(key, value)

__getattr__(item)

Получение атрибута.

Source code in src/qtasks/configs/config.py 
64
65
66
67
68
69
70
def __getattr__(self, item):
    """Получение атрибута."""
    # Проверяем динамические поля
    if item in self._dynamic_fields:
        return self._dynamic_fields[item]
    # Вызов стандартного поведения dataclass
    raise AttributeError(f"{type(self).__name__} has no attribute '{item}'")

__setattr__(key, value)

Установка атрибута.

Source code in src/qtasks/configs/config.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def __setattr__(self, key, value):
    """Установка атрибута."""
    # Для полей dataclass
    if key in self.__annotations__:
        object.__setattr__(self, key, value)
        if (
            "_callbacks" in self.__dict__
        ):  # Проверяем наличие для избежания рекурсии
            self._notify(key, value)
    # Для внутренних атрибутов
    elif key.startswith("_"):
        object.__setattr__(self, key, value)
    # Для динамических полей
    else:
        dynamic_fields = object.__getattribute__(self, "_dynamic_fields")
        dynamic_fields[key] = value
        self._notify(key, value)

subscribe(callback)

Подписка на изменение.

Source code in src/qtasks/configs/config.py 
56
57
58
def subscribe(self, callback: Callable[["QueueConfig", str, Any], None]):
    """Подписка на изменение."""
    self._callbacks.append(callback)