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

BaseGlobalConfig - пишем свой GlobalConfig

Base Configurations.

BaseGlobalConfig

Bases: ABC

BaseGlobalConfig - Абстрактный класс, который является фундаментом для Глобального Конфига.

Пример

from qtasks import QueueTasks
from qtasks.configs.base import BaseGlobalConfig

class MyGlobalConfig(BaseGlobalConfig):
    def __init__(self, name: str = None):
        super().__init__(name=name)
        pass
Source code in src/qtasks/configs/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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class BaseGlobalConfig(ABC):
    """
    `BaseGlobalConfig` - Абстрактный класс, который является фундаментом для Глобального Конфига.

    ## Пример

    ```python
    from qtasks import QueueTasks
    from qtasks.configs.base import BaseGlobalConfig

    class MyGlobalConfig(BaseGlobalConfig):
        def __init__(self, name: str = None):
            super().__init__(name=name)
            pass
    ```
    """

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

                    По умолчанию: `None`.
                    """
            ),
        ] = 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(
                """
                    События.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
    ):
        """Инициализация контекста.

        Args:
            name (str, optional): Имя проекта. По умолчанию: `None`.
            log (Logger, optional): Логгер. По умолчанию: `None`.
            config (QueueConfig, optional): Конфигурация. По умолчанию: `None`.
            events (BaseEvents, optional): События. По умолчанию: `None`.
        """
        self.name = name
        self.config = config or QueueConfig()
        self.log = (
            log.with_subname("GlobalConfig")
            if log
            else Logger(
                name=self.name,
                subname="GlobalConfig",
                default_level=self.config.logs_default_level_server,
                format=self.config.logs_format,
            )
        )
        self.events = events
        self.client = None
        self.plugins: Dict[str, List["BasePlugin"]] = {}

        self.init_plugins()

    @abstractmethod
    def set(self, **kwargs) -> None:
        """Добавить новое значение.

        Args:
            kwargs (dict, optional): kwags задачи. По умолчанию `{}`.
        """
        pass

    @abstractmethod
    def get(self, key: str, name: str) -> Any:
        """Получить значение.

        Args:
            key (str): Ключ.
            name (str): Имя.

        Returns:
            Any: Значение.
        """
        pass

    @abstractmethod
    def get_all(self, key: str) -> Union[dict, list, tuple]:
        """Получить все значения.

        Args:
            key (str): Ключ.

        Returns:
            Dict[str, Any] | List[Any] | Tuple[Any]: Значения.
        """
        pass

    @abstractmethod
    def get_match(self, match: str) -> Union[Any, dict, list, tuple]:
        """Получить значения по паттерну.

        Args:
            match (str): Паттерн.

        Returns:
            Any | Dict[str, Any] | List[Any] | Tuple[Any]: Значение или Значения.
        """
        pass

    @abstractmethod
    def start(self) -> None:
        """Запуск Брокера. Эта функция задействуется основным экземпляром `QueueTasks` через `run_forever."""
        pass

    @abstractmethod
    def stop(self) -> None:
        """Останавливает Глобальный Конфиг. Эта функция задействуется основным экземпляром `QueueTasks` после завершения функции `run_forever."""
        pass

    def update_config(
        self,
        config: Annotated[
            QueueConfig,
            Doc(
                """
                    Конфиг.
                    """
            ),
        ],
    ) -> None:
        """Обновляет конфиг брокера.

        Args:
            config (QueueConfig): Конфиг.
        """
        self.config = config
        return

    def add_plugin(
        self,
        plugin: Annotated[
            "BasePlugin",
            Doc(
                """
                    Плагин.
                    """
            ),
        ],
        trigger_names: Annotated[
            Optional[List[str]],
            Doc(
                """
                    Имя триггеров для плагина.

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

        Args:
            plugin (BasePlugin): Плагин
            trigger_names (List[str], optional): Имя триггеров для плагина. По умолчанию: будет добавлен в `Globals`.
        """
        trigger_names = trigger_names or ["Globals"]

        for name in trigger_names:
            if name not in self.plugins:
                self.plugins.update({name: [plugin]})
            else:
                self.plugins[name].append(plugin)
        return

    def init_plugins(self):
        """Инициализация плагинов."""
        pass

__init__(name=None, log=None, config=None, events=None)

Инициализация контекста.

Parameters:

Name Type Description Default
name str

Имя проекта. По умолчанию: None.

 None
log Logger

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

 None
config QueueConfig

Конфигурация. По умолчанию: None.

 None
events BaseEvents

События. По умолчанию: None.

 None
Source code in src/qtasks/configs/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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __init__(
    self,
    name: Annotated[
        Optional[str],
        Doc(
            """
                Имя проекта. Это имя можно использовать для тегов для GlobalConfig.

                По умолчанию: `None`.
                """
        ),
    ] = 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(
            """
                События.

                По умолчанию: `None`.
                """
        ),
    ] = None,
):
    """Инициализация контекста.

    Args:
        name (str, optional): Имя проекта. По умолчанию: `None`.
        log (Logger, optional): Логгер. По умолчанию: `None`.
        config (QueueConfig, optional): Конфигурация. По умолчанию: `None`.
        events (BaseEvents, optional): События. По умолчанию: `None`.
    """
    self.name = name
    self.config = config or QueueConfig()
    self.log = (
        log.with_subname("GlobalConfig")
        if log
        else Logger(
            name=self.name,
            subname="GlobalConfig",
            default_level=self.config.logs_default_level_server,
            format=self.config.logs_format,
        )
    )
    self.events = events
    self.client = None
    self.plugins: Dict[str, List["BasePlugin"]] = {}

    self.init_plugins()

add_plugin(plugin, trigger_names=None)

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

Parameters:

Name Type Description Default
plugin BasePlugin

Плагин

 required
trigger_names List[str]

Имя триггеров для плагина. По умолчанию: будет добавлен в Globals.

 None
Source code in src/qtasks/configs/base.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def add_plugin(
    self,
    plugin: Annotated[
        "BasePlugin",
        Doc(
            """
                Плагин.
                """
        ),
    ],
    trigger_names: Annotated[
        Optional[List[str]],
        Doc(
            """
                Имя триггеров для плагина.

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

    Args:
        plugin (BasePlugin): Плагин
        trigger_names (List[str], optional): Имя триггеров для плагина. По умолчанию: будет добавлен в `Globals`.
    """
    trigger_names = trigger_names or ["Globals"]

    for name in trigger_names:
        if name not in self.plugins:
            self.plugins.update({name: [plugin]})
        else:
            self.plugins[name].append(plugin)
    return

get(key, name) abstractmethod

Получить значение.

Parameters:

Name Type Description Default
key str

Ключ.

 required
name str

Имя.

 required

Returns:

Name Type Description
Any Any

Значение.
Source code in src/qtasks/configs/base.py 
110
111
112
113
114
115
116
117
118
119
120
121
@abstractmethod
def get(self, key: str, name: str) -> Any:
    """Получить значение.

    Args:
        key (str): Ключ.
        name (str): Имя.

    Returns:
        Any: Значение.
    """
    pass

get_all(key) abstractmethod

Получить все значения.

Parameters:

Name Type Description Default
key str

Ключ.

 required

Returns:

Type Description
Union[dict, list, tuple]

Dict[str, Any] | List[Any] | Tuple[Any]: Значения.
Source code in src/qtasks/configs/base.py 
123
124
125
126
127
128
129
130
131
132
133
@abstractmethod
def get_all(self, key: str) -> Union[dict, list, tuple]:
    """Получить все значения.

    Args:
        key (str): Ключ.

    Returns:
        Dict[str, Any] | List[Any] | Tuple[Any]: Значения.
    """
    pass

get_match(match) abstractmethod

Получить значения по паттерну.

Parameters:

Name Type Description Default
match str

Паттерн.

 required

Returns:

Type Description
Union[Any, dict, list, tuple]

Any | Dict[str, Any] | List[Any] | Tuple[Any]: Значение или Значения.
Source code in src/qtasks/configs/base.py 
135
136
137
138
139
140
141
142
143
144
145
@abstractmethod
def get_match(self, match: str) -> Union[Any, dict, list, tuple]:
    """Получить значения по паттерну.

    Args:
        match (str): Паттерн.

    Returns:
        Any | Dict[str, Any] | List[Any] | Tuple[Any]: Значение или Значения.
    """
    pass

init_plugins()

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

Source code in src/qtasks/configs/base.py 
212
213
214
def init_plugins(self):
    """Инициализация плагинов."""
    pass

set(**kwargs) abstractmethod

Добавить новое значение.

Parameters:

Name Type Description Default
kwargs dict

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

 {}
Source code in src/qtasks/configs/base.py 
101
102
103
104
105
106
107
108
@abstractmethod
def set(self, **kwargs) -> None:
    """Добавить новое значение.

    Args:
        kwargs (dict, optional): kwags задачи. По умолчанию `{}`.
    """
    pass

start() abstractmethod

Запуск Брокера. Эта функция задействуется основным экземпляром QueueTasks через `run_forever.

Source code in src/qtasks/configs/base.py 
147
148
149
150
@abstractmethod
def start(self) -> None:
    """Запуск Брокера. Эта функция задействуется основным экземпляром `QueueTasks` через `run_forever."""
    pass

stop() abstractmethod

Останавливает Глобальный Конфиг. Эта функция задействуется основным экземпляром QueueTasks после завершения функции `run_forever.

Source code in src/qtasks/configs/base.py 
152
153
154
155
@abstractmethod
def stop(self) -> None:
    """Останавливает Глобальный Конфиг. Эта функция задействуется основным экземпляром `QueueTasks` после завершения функции `run_forever."""
    pass

update_config(config)

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

Parameters:

Name Type Description Default
config QueueConfig

Конфиг.

 required
Source code in src/qtasks/configs/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def update_config(
    self,
    config: Annotated[
        QueueConfig,
        Doc(
            """
                Конфиг.
                """
        ),
    ],
) -> None:
    """Обновляет конфиг брокера.

    Args:
        config (QueueConfig): Конфиг.
    """
    self.config = config
    return