py.lib

Yohn Y. 2022-08-18 Parent:84b54a8a6d4c Child:f1a05e880961

35:ab4cf9f4f10a 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, Callable
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 def difficult_type_recognizer(type_obj, value):
22 """\
23 Магия борющаяся с костылями type hinting
24 """
25 if type(type_obj).__name__ == '_GenericAlias':
26 if (
27 hasattr(type_obj, '__args__')
28 and type(type_obj.__args__) == tuple
29 and type_obj.__args__
30 ):
31 if type_obj.__name__ == 'List':
32 return list(map(type_obj.__args__[0], value))
33
34 elif type_obj.__name__ == 'Tuple':
35 return tuple(map(type_obj.__args__[0], value))
36
37 else:
38 raise ValueError('Неизвестный тип')
39
40 else:
41 ValueError('Неизвестный тип')
42
43 elif type(type_obj).__name__ == '_UnionGenericAlias':
44 if type_obj.__name__ not in ('Union', 'Optional'):
45 raise TypeError(f'Неизвестный подтип _UnionGenericAlias: {type_obj.__name__}')
46
47 if not (
48 hasattr(type_obj, '__args__')
49 and type(type_obj.__args__) == tuple
50 and type_obj.__args__
51 ):
52 raise TypeError(f'Не ясно как работать с типом не вижу аргументов: {type_obj.__name__}')
53
54 for _t in type_obj.__args__:
55 if _t.__name__ == 'NoneType':
56 continue
57
58 try:
59 return _t(value)
60
61 except (TypeError, ValueError):
62 continue
63
64 raise ValueError('Не удалось привести значение к типу')
65
66
67 class CPHSectionBase:
68 """\
69 Базовый класс обработки секции конфигурационного файла
70 """
71
72 def get(self, config_prop_name: str, dc_prop_name: str):
73 """\
74 Получить свойство из конфигурационного файла
75 """
76 raise NotImplemented()
77
78 def __enter__(self):
79 return self
80
81 def __exit__(self, exc_type, exc_val, exc_tb):
82 raise NotImplemented()
83
84
85 class CPHAbsentSection(CPHSectionBase):
86 """\
87 Класс создаваемый на отсутствующую секцию конфигурационного файла
88 """
89 def get(self, config_prop_name: str, dc_prop_name: str):
90 raise NoSectionNotification()
91
92 def __exit__(self, exc_type, exc_val, exc_tb):
93 if exc_type == NoSectionNotification:
94 return True
95
96
97 class CPHParamGetter(CPHSectionBase):
98 def __init__(self, parser_helper_object):
99 self.ph = parser_helper_object
100 self.params = {}
101
102 def _add_param(self, param_name: str, param_val: Any, parser: Optional[Callable[[Any], Any]] = None):
103 """\
104 Непосредственное добавление полученного параметра со всеми проверками.
105 """
106
107 if parser is not None and param_val is not None:
108 param_val = parser(param_val)
109
110 fld = self.ph.fields.get(param_name)
111 if not isinstance(fld, Field):
112 raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '
113 f'которое мы должны заполнить из параметра конфигурации: {fld}')
114
115 if param_val is not None:
116 try:
117 res = fld.type(param_val)
118
119 except (ValueError, TypeError) as e:
120 try:
121 res = difficult_type_recognizer(fld.type, param_val)
122
123 except (ValueError, TypeError):
124 raise ConfigParseHelperError(f'При приведении параметра к '
125 f'заданному типу произошла ошибка: '
126 f'значение="{param_val}" ошибка="{e}"')
127
128 else:
129 if fld.default is not MISSING:
130 res = fld.default
131
132 elif fld.default_factory is not MISSING:
133 res = fld.default_factory()
134
135 else:
136 raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')
137
138 self.params[param_name] = res
139
140 def __exit__(self, exc_type, exc_val, exc_tb):
141 if exc_type is None:
142 self.ph.add_params(self.params)
143
144 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
145 raise NoSectionNotification()
146
147
148 class CPHSection(CPHParamGetter):
149 """\
150 Класс производящий разбор конкретной секции конфигурации
151 """
152
153 def __init__(self, parser_helper_object, section: str):
154 super().__init__(parser_helper_object)
155 self.section_name = section
156 self.section = parser_helper_object.conf_parser[section]
157
158 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
159 """\
160 :param config_prop_name: Имя опции в файле конфигурации
161 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
162 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
163 """
164 try:
165 self._add_param(dc_prop_name, self.section.get(config_prop_name), parser)
166
167 except ConfigParseHelperError as e:
168 raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '
169 f'в секции "{self.section_name}": {e}')
170
171
172 class CPHEnvParser(CPHParamGetter):
173 """\
174 Класс для разбора переменных окружения в том же ключе, что и файла конфигурации
175 """
176
177 def get(self, env_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
178 """\
179 :param env_name: Имя переменной окружения, хранящей опцию
180 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
181 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
182 """
183
184 try:
185 self._add_param(dc_prop_name, getenv(env_name), parser)
186
187 except ConfigParseHelperError as e:
188 raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')
189
190
191 class CPHObjectsListGetter:
192 """\
193 Помощник для случаев, когда в наборе секций хранится однотипный набор объектов
194 """
195 def __init__(self, config_object_class: type, config_parser: ConfigParser, sections: Iterable[str]):
196 self.sections = sections
197 self.conf_parser = config_parser
198
199 if not is_dataclass(config_object_class):
200 raise ConfigParseHelperError(f'Представленный в качестве представления объекта конфигурации '
201 f'класс не является классом данных: {config_object_class.__name__}')
202
203 self.res_obj = config_object_class
204 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
205 self.obj_list = []
206 self.ident_list = []
207
208 def add_params(self, params: Dict[str, Any]):
209 try:
210 self.obj_list.append(self.res_obj(**params))
211
212 except (ValueError, TypeError) as e:
213 raise ConfigParseHelperError(f'Ошибка создания объекта объекта конфигурации, '
214 f'списка объектов конфигурации: {e}')
215
216 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
217 """\
218 Подготавливаем список соответствия названий параметров в секции конкретным свойствам данного
219 в помощник класса
220
221 :param config_prop_name: Имя опции в файле конфигурации
222 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
223 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
224 """
225
226 self.ident_list.append((config_prop_name, dc_prop_name, parser))
227
228 def get_config_objects(self) -> List[object]:
229 for section in self.sections:
230 try:
231 with CPHSection(self, section) as section_helper:
232 for conf_prop, dc_prop, parser in self.ident_list:
233 section_helper.get(conf_prop, dc_prop, parser)
234
235 except ConfigParseHelperError as e:
236 raise ConfigParseHelperError(f'Ошибка при разборе секции "{section}": {e}')
237
238 res = self.obj_list[:]
239 self.obj_list.clear()
240
241 return res
242
243
244 class ConfigParseHelper:
245 """\
246 Помощник разбора конфигурации
247 """
248 def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):
249 """\
250 :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации
251 :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле
252 """
253
254 if required_sections is not None:
255 self.req_sections = set(required_sections)
256
257 else:
258 self.req_sections = set()
259
260 if not is_dataclass(config_object_class):
261 raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '
262 f'классом данных: {config_object_class.__name__}')
263
264 self.res_obj = config_object_class
265 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
266 self.conf_parser: Optional[ConfigParser] = None
267 self.config_params = {}
268 self.config_params_lock = RLock()
269
270 def add_params(self, params: Dict[str, Any]):
271 self.config_params_lock.acquire()
272 try:
273 self.config_params.update(params)
274
275 finally:
276 self.config_params_lock.release()
277
278 def section(self, section_name: str) -> CPHSectionBase:
279 if self.conf_parser is None:
280 raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')
281
282 if self.conf_parser.has_section(section_name):
283 return CPHSection(self, section_name)
284
285 else:
286 return CPHAbsentSection()
287
288 def load(self, filename: str):
289 res = ConfigParser()
290 try:
291 res.read(filename)
292
293 except (TypeError, IOError, OSError, ValueError) as e:
294 raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '
295 f'ошибка="{e}"')
296
297 missing_sections = self.req_sections - set(res.sections())
298
299 if missing_sections:
300 missing_sections = ', '.join(missing_sections)
301 raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')
302
303 self.conf_parser = res
304
305 def get_config(self):
306 try:
307 return self.res_obj(**self.config_params)
308
309 except (ValueError, TypeError) as e:
310 raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')
311
312 def get_sections(self):
313 return self.conf_parser.sections()