py.lib

Yohn Y. 2022-08-13 Parent:4186c3b229fa Child:57f63bf31fd8

32:eb41cc498ddb Go to Latest

py.lib/config_parse_helper.py

+ Поддержали наборы объектов концигурации в библиотеке помощника разбора конфигурации + Позволяем в помощнике получить секции конфигурации.

Download raw file
View source Diff to previous Annotate
History
1 # coding: utf-8
2 from configparser import ConfigParser
3 from dataclasses import is_dataclass, fields, Field, MISSING
4 from typing import Iterable, Optional, Dict, Any, List
5 from threading import RLock
6 from os import getenv
7
8
9 class ConfigParseHelperError(Exception):
10 """\
11 Ошибки в модуле помощника разборщика конфигураций
12 """
13
14
15 class NoSectionNotification(ConfigParseHelperError):
16 """\
17 Оповещение об отсутствующей секции конфигурации
18 """
19
20
21 class CPHSectionBase:
22 """\
23 Базовый класс обработки секции конфигурационного файла
24 """
25
26 def get(self, config_prop_name: str, dc_prop_name: str):
27 """\
28 Получить свойство из конфигурационного файла
29 """
30 raise NotImplemented()
31
32 def __enter__(self):
33 return self
34
35 def __exit__(self, exc_type, exc_val, exc_tb):
36 raise NotImplemented()
37
38
39 class CPHAbsentSection(CPHSectionBase):
40 """\
41 Класс создаваемый на отсутствующую секцию конфигурационного файла
42 """
43 def get(self, config_prop_name: str, dc_prop_name: str):
44 raise NoSectionNotification()
45
46 def __exit__(self, exc_type, exc_val, exc_tb):
47 if exc_type == NoSectionNotification:
48 return True
49
50
51 class CPHParamGetter(CPHSectionBase):
52 def __init__(self, parser_helper_object):
53 self.ph = parser_helper_object
54 self.params = {}
55
56 def _add_param(self, param_name: str, param_val: Any):
57 """\
58 Непосредственное добавление полученного параметра со всеми проверками.
59 """
60
61 fld = self.ph.fields.get(param_name)
62 if not isinstance(fld, Field):
63 raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '
64 f'которое мы должны заполнить из параметра конфигурации: {fld}')
65
66 if param_val is not None:
67 try:
68 res = fld.type(param_val)
69
70 except (ValueError, TypeError) as e:
71 raise ConfigParseHelperError(f'При приведении параметра к '
72 f'заданному типу произошла ошибка: '
73 f'значение="{param_val}" ошибка="{e}"')
74
75 else:
76 if fld.default is not MISSING:
77 res = fld.default
78
79 elif fld.default_factory is not MISSING:
80 res = fld.default_factory()
81
82 else:
83 raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')
84
85 self.params[param_name] = res
86
87 def __exit__(self, exc_type, exc_val, exc_tb):
88 if exc_type is None:
89 self.ph.add_params(self.params)
90
91 def get(self, config_prop_name: str, dc_prop_name: str):
92 raise NoSectionNotification()
93
94
95 class CPHSection(CPHParamGetter):
96 """\
97 Класс производящий разбор конкретной секции конфигурации
98 """
99
100 def __init__(self, parser_helper_object, section: str):
101 super().__init__(parser_helper_object)
102 self.section_name = section
103 self.section = parser_helper_object.conf_parser[section]
104
105 def get(self, config_prop_name: str, dc_prop_name: str):
106 """\
107 :param config_prop_name: Имя опции в файле конфигурации
108 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
109 """
110 try:
111 self._add_param(dc_prop_name, self.section.get(config_prop_name))
112
113 except ConfigParseHelperError as e:
114 raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '
115 f'в секции "{self.section_name}": {e}')
116
117
118 class CPHEnvParser(CPHParamGetter):
119 """\
120 Класс для разбора переменных окружения в том же ключе, что и файла конфигурации
121 """
122
123 def get(self, env_name: str, dc_prop_name: str):
124 """\
125 :param env_name: Имя переменной окружения, хранящей опцию
126 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
127 """
128
129 try:
130 self._add_param(dc_prop_name, getenv(env_name))
131
132 except ConfigParseHelperError as e:
133 raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')
134
135
136 class CPHObjectsListGetter:
137 """\
138 Помощник для случаев, когда в наборе секций хранится однотипный набор объектов
139 """
140 def __init__(self, config_object_class: type, config_parser: ConfigParser, sections: Iterable[str]):
141 self.sections = sections
142 self.conf_parser = config_parser
143
144 if not is_dataclass(config_object_class):
145 raise ConfigParseHelperError(f'Представленный в качестве представления объекта конфигурации '
146 f'класс не является классом данных: {config_object_class.__name__}')
147
148 self.res_obj = config_object_class
149 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
150 self.obj_list = []
151 self.ident_list = []
152
153 def add_params(self, params: Dict[str, Any]):
154 try:
155 self.obj_list.append(self.res_obj(**params))
156
157 except (ValueError, TypeError) as e:
158 raise ConfigParseHelperError(f'Ошибка создания объекта объекта конфигурации, '
159 f'списка объектов конфигурации: {e}')
160
161 def get(self, config_prop_name: str, dc_prop_name: str):
162 """\
163 Подготавливаем список соответствия названий параметров в секции конкретным свойствам данного
164 в помощник класса
165
166 :param config_prop_name: Имя опции в файле конфигурации
167 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
168 """
169
170 self.ident_list.append((config_prop_name, dc_prop_name))
171
172 def get_config_objects(self) -> List[object]:
173 for section in self.sections:
174 try:
175 with CPHSection(self, section) as section_helper:
176 for conf_prop, dc_prop in self.ident_list:
177 section_helper.get(conf_prop, dc_prop)
178
179 except ConfigParseHelperError as e:
180 raise ConfigParseHelperError(f'Ошибка при разборе секции "{section}": {e}')
181
182 res = self.obj_list[:]
183 self.obj_list.clear()
184
185 return res
186
187
188 class ConfigParseHelper:
189 """\
190 Помощник разбора конфигурации
191 """
192 def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):
193 """\
194 :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации
195 :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле
196 """
197
198 if required_sections is not None:
199 self.req_sections = set(required_sections)
200
201 else:
202 self.req_sections = set()
203
204 if not is_dataclass(config_object_class):
205 raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '
206 f'классом данных: {config_object_class.__name__}')
207
208 self.res_obj = config_object_class
209 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
210 self.conf_parser: Optional[ConfigParser] = None
211 self.config_params = {}
212 self.config_params_lock = RLock()
213
214 def add_params(self, params: Dict[str, Any]):
215 self.config_params_lock.acquire()
216 try:
217 self.config_params.update(params)
218
219 finally:
220 self.config_params_lock.release()
221
222 def section(self, section_name: str) -> CPHSectionBase:
223 if self.conf_parser is None:
224 raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')
225
226 if self.conf_parser.has_section(section_name):
227 return CPHSection(self, section_name)
228
229 else:
230 return CPHAbsentSection()
231
232 def load(self, filename: str):
233 res = ConfigParser()
234 try:
235 res.read(filename)
236
237 except (TypeError, IOError, OSError, ValueError) as e:
238 raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '
239 f'ошибка="{e}"')
240
241 missing_sections = self.req_sections - set(res.sections())
242
243 if missing_sections:
244 missing_sections = ', '.join(missing_sections)
245 raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')
246
247 self.conf_parser = res
248
249 def get_config(self):
250 try:
251 return self.res_obj(**self.config_params)
252
253 except (ValueError, TypeError) as e:
254 raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')
255
256 def get_sections(self):
257 return self.conf_parser.sections()