py.lib

Yohn Y. 2022-08-25 Parent:4f4cc2fc9805 Child:366c9fe26d76

39:b8d559c989d6 Go to Latest

py.lib/type_utils/config_parse_helper.py

+ Возможность воссоздавать объекты классов из кортежей и словарей

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