py.lib

Yohn Y. 2022-08-19 Parent:f1a05e880961

37:ae0107755941 Go to Latest

py.lib/config_parse_helper.py

. Исправление ошибок и рефакторинг

Download raw file
View source Diff to previous Annotate
History
awgur@31 1 # coding: utf-8
awgur@31 2 from configparser import ConfigParser
awgur@31 3 from dataclasses import is_dataclass, fields, Field, MISSING
awgur@36 4 from typing import Iterable, Optional, Dict, Any, List, Callable, Union
awgur@31 5 from threading import RLock
awgur@31 6 from os import getenv
awgur@31 7
awgur@31 8
awgur@31 9 class ConfigParseHelperError(Exception):
awgur@31 10 """\
awgur@31 11 Ошибки в модуле помощника разборщика конфигураций
awgur@31 12 """
awgur@31 13
awgur@31 14
awgur@31 15 class NoSectionNotification(ConfigParseHelperError):
awgur@31 16 """\
awgur@31 17 Оповещение об отсутствующей секции конфигурации
awgur@31 18 """
awgur@31 19
awgur@31 20
awgur@36 21 class TypeDescriber:
awgur@33 22 """\
awgur@36 23 Реализует паттерн "адаптер" поверх типов, для упрощения приведения значений типов к объявленным формам
awgur@33 24 """
awgur@36 25 def __init__(self, name, cast, like, is_complex=False):
awgur@36 26 self.__name__ = name
awgur@36 27 self.cast = cast
awgur@36 28 self.like = like
awgur@36 29 self.is_complex = is_complex
awgur@36 30
awgur@37 31 def check(self, instance):
awgur@36 32 if self.like is None:
awgur@36 33 return False
awgur@33 34
awgur@36 35 else:
awgur@37 36 return check_instance(instance, self.like)
awgur@33 37
awgur@36 38 def __repr__(self):
awgur@36 39 return f'<TypeDescriber({self.__name__}, {self.like})>'
awgur@36 40
awgur@36 41 def __call__(self, val):
awgur@36 42 if val is None:
awgur@36 43 return None
awgur@33 44
awgur@37 45 elif not self.is_complex and check_instance(val, self.like):
awgur@37 46 return val
awgur@37 47
awgur@33 48 else:
awgur@37 49 return self.cast(val)
awgur@37 50
awgur@37 51
awgur@37 52 def check_instance(obj, types):
awgur@37 53 if types is None:
awgur@37 54 return False
awgur@37 55
awgur@37 56 elif isinstance(types, (tuple, list)):
awgur@37 57 _flag = False
awgur@37 58 for t in types:
awgur@37 59 if check_instance(obj, t):
awgur@37 60 _flag = True
awgur@37 61 break
awgur@37 62
awgur@37 63 return _flag
awgur@37 64
awgur@37 65 elif isinstance(types, TypeDescriber):
awgur@37 66 return types.check(obj)
awgur@37 67
awgur@37 68 else:
awgur@37 69 return isinstance(obj, types)
awgur@36 70
awgur@36 71
awgur@36 72 def cast_iterator(t: Union[type, TypeDescriber], lst: Iterable):
awgur@36 73 """\
awgur@36 74 Обрабатывает последовательности единого типа.
awgur@36 75 """
awgur@36 76 for i in lst:
awgur@37 77 if check_instance(i, t):
awgur@36 78 yield i
awgur@36 79
awgur@37 80 else:
awgur@37 81 try:
awgur@37 82 yield t(i)
awgur@36 83
awgur@37 84 except (TypeError, ValueError) as e:
awgur@37 85 raise ValueError(f'Не удалось привести значение к нужному типу: '
awgur@37 86 f'тип={t.__name__}; знач={i}')
awgur@36 87
awgur@36 88
awgur@36 89 def multi_item_tuple(tt, val):
awgur@36 90 """\
awgur@36 91 Обрабатывает кортежи, состоящие из нескольких значений типов (кортежи элементов разных типов)
awgur@36 92 :param tt: Последовательность составляющих кортеж типов
awgur@36 93 :param val: итерируемый объект, хранящий значения в указанном порядке.
awgur@36 94 """
awgur@36 95 val = list(val)
awgur@36 96 t_len = len(tt)
awgur@36 97 if t_len == 0:
awgur@36 98 raise ValueError('При вызове процедуры конвертации котежей, не были указаны типы')
awgur@36 99
awgur@36 100 if len(val) != t_len:
awgur@36 101 raise ValueError(f'Значение не содержит положенных {t_len} элементов: {len(val)} - {val}')
awgur@36 102
awgur@36 103 res = []
awgur@36 104
awgur@36 105 for i in range(t_len):
awgur@37 106 if check_instance(val[i], tt[i]):
awgur@37 107 res.append(val[i])
awgur@36 108
awgur@37 109 else:
awgur@37 110 try:
awgur@37 111 res.append(tt[i](val[i]))
awgur@36 112
awgur@37 113 except (TypeError, ValueError) as e:
awgur@37 114 raise ValueError(f'Не удалось привести значение к нужному типу: '
awgur@37 115 f'тип={tt[i].__name__}; знач={val[i]}')
awgur@36 116
awgur@36 117 return tuple(res)
awgur@36 118
awgur@33 119
awgur@36 120 def union_processor(tt, val):
awgur@36 121 """\
awgur@36 122 Пытается привести значение к одному из указанных типов
awgur@36 123 :param tt: список возможных типов
awgur@36 124 :param val: приводимое значение
awgur@36 125 """
awgur@36 126 if val is None:
awgur@36 127 return val
awgur@36 128
awgur@36 129 res = None
awgur@36 130 ex = []
awgur@36 131
awgur@36 132 if len(tt) == 0:
awgur@36 133 raise ValueError('Не указан ни один тип в составном типе Union')
awgur@36 134
awgur@37 135 sorted_types_begin = [ t for t in tt if t in (int, float, bool)]
awgur@37 136 sorted_types_body = [ t for t in tt if t not in (int, float, bool, str)]
awgur@37 137 sorted_types_end = [ t for t in tt if t == str]
awgur@37 138
awgur@37 139 for t in sorted_types_begin + sorted_types_body + sorted_types_end:
awgur@37 140 _t = get_type_describer(t)
awgur@36 141 try:
awgur@37 142 res = _t(val)
awgur@36 143 break
awgur@36 144
awgur@36 145 except (TypeError, ValueError) as e:
awgur@36 146 ex.append(f'{t}: {e}')
awgur@36 147
awgur@36 148 if res is None:
awgur@36 149 raise ValueError('Не удалось привести значение не к одному из типов:\n' + '\n'.join(ex))
awgur@36 150
awgur@36 151 else:
awgur@36 152 return res
awgur@36 153
awgur@36 154
awgur@36 155 def dict_processor(tt, val):
awgur@36 156 if len(tt) != 2:
awgur@36 157 raise ValueError(f'Попытка воссоздать словарь со странным количеством аргументов типа: {tt}')
awgur@36 158
awgur@36 159 try:
awgur@36 160 _d = dict(val)
awgur@36 161
awgur@36 162 except (TypeError, ValueError) as e:
awgur@36 163 raise ValueError(f'Не удалось воссоздать словарь из представленного значения: {e}')
awgur@36 164
awgur@36 165 _p = []
awgur@36 166
awgur@36 167 for k, v in _d.items():
awgur@36 168 try:
awgur@36 169 _p.append((tt[0](k), tt[1](v)))
awgur@36 170
awgur@36 171 except (TypeError, ValueError) as e:
awgur@36 172 raise ValueError(f'Не удалось привести значения элемента словаря к требуемому типу: '
awgur@36 173 f'key="{k}" value="{v}"')
awgur@33 174
awgur@36 175 return dict(_p)
awgur@36 176
awgur@36 177
awgur@36 178 def get_type_describer(t) -> TypeDescriber:
awgur@36 179 if type(t).__name__ == '_GenericAlias':
awgur@36 180 try:
awgur@36 181 _args = t.__args__
awgur@36 182
awgur@36 183 except AttributeError:
awgur@36 184 raise TypeError(f'Неизвестный тип хранения внутренних типов для представителя сложного '
awgur@36 185 f'типа "_GenericAlias": {t}')
awgur@36 186
awgur@36 187 if t.__name__ == 'List':
awgur@36 188 try:
awgur@36 189 _t = _args[0]
awgur@36 190
awgur@36 191 except IndexError:
awgur@36 192 raise ValueError(f'Тип {t} не содержит в себе типа своих элементов')
awgur@36 193
awgur@36 194 return TypeDescriber(
awgur@36 195 name=f'{t.__name__}[{_t.__name__}]',
awgur@36 196 cast=lambda x: list(cast_iterator(get_type_describer(_t).cast, x)),
awgur@36 197 like=list,
awgur@36 198 is_complex=True
awgur@36 199 )
awgur@36 200
awgur@36 201 elif t.__name__ == 'Tuple':
awgur@36 202 if not _args:
awgur@36 203 raise ValueError(f'Тип {t} не содержит в себе пояснений по составу хранящихся в нём типов')
awgur@36 204
awgur@36 205 if len(_args) == 1:
awgur@36 206 _t = _args[0]
awgur@36 207
awgur@36 208 return TypeDescriber(
awgur@36 209 name=f'Tuple[{_t.__name__}]',
awgur@36 210 cast=lambda x: list(cast_iterator(get_type_describer(_t).cast, x)),
awgur@36 211 like=tuple,
awgur@36 212 is_complex=True
awgur@36 213 )
awgur@36 214
awgur@36 215 else:
awgur@36 216 _name = ', '.join(map(lambda x: x.__name__, _args))
awgur@36 217 _cast_args = tuple(get_type_describer(i) for i in _args)
awgur@33 218
awgur@36 219 return TypeDescriber(
awgur@36 220 name=f'Tuple[{_name}]',
awgur@36 221 cast=lambda x: multi_item_tuple(_cast_args, x),
awgur@36 222 like=tuple,
awgur@36 223 is_complex=True
awgur@36 224 )
awgur@36 225
awgur@36 226 elif t.__name__ == 'Dict':
awgur@36 227 if len(_args) != 2:
awgur@36 228 raise ValueError(f'Неожиданное количество значений типа в составном типе Dict: '
awgur@36 229 f'{len(_args)} values=({_args})')
awgur@36 230
awgur@36 231 _name = ', '.join(map(lambda x: x.__name__, _args))
awgur@36 232
awgur@36 233 return TypeDescriber(
awgur@36 234 name=f'Dict[{_name}]',
awgur@36 235 cast=lambda x: dict_processor(_args, x),
awgur@36 236 like=dict,
awgur@36 237 is_complex=True
awgur@36 238 )
awgur@36 239
awgur@36 240 else:
awgur@36 241 raise ValueError(f'Неизвестный представитель типа "_GenericAlias": {t.__name__}')
awgur@33 242
awgur@36 243 elif type(t).__name__ == '_UnionGenericAlias':
awgur@36 244 if t.__name__ not in ('Union', 'Optional'):
awgur@36 245 raise TypeError(f'Неизвестный подтип _UnionGenericAlias: {t.__name__}')
awgur@36 246
awgur@36 247 try:
awgur@36 248 _args = t.__args__
awgur@36 249
awgur@36 250 except AttributeError:
awgur@36 251 raise TypeError(f'Неизвестный тип хранения внутренних типов для представителя сложного '
awgur@36 252 f'типа "_UnionGenericAlias": {t}')
awgur@36 253
awgur@36 254 if len(_args) == 0:
awgur@36 255 raise ValueError('Не указан ни один тип в конструкции Union')
awgur@33 256
awgur@36 257 _cast_args = tuple(map(get_type_describer, _args))
awgur@36 258 _args_name = ', '.join(map(lambda x: x.__name__, _args))
awgur@33 259
awgur@36 260 return TypeDescriber(
awgur@36 261 name=f'Union[{_args_name}]',
awgur@36 262 cast=lambda x: union_processor(_cast_args, x),
awgur@36 263 like=None
awgur@36 264 )
awgur@36 265
awgur@36 266 else:
awgur@36 267 return TypeDescriber(
awgur@36 268 name=t.__name__,
awgur@36 269 cast=t,
awgur@36 270 like=t
awgur@36 271 )
awgur@33 272
awgur@33 273
awgur@31 274 class CPHSectionBase:
awgur@31 275 """\
awgur@31 276 Базовый класс обработки секции конфигурационного файла
awgur@31 277 """
awgur@31 278
awgur@31 279 def get(self, config_prop_name: str, dc_prop_name: str):
awgur@31 280 """\
awgur@31 281 Получить свойство из конфигурационного файла
awgur@31 282 """
awgur@31 283 raise NotImplemented()
awgur@31 284
awgur@31 285 def __enter__(self):
awgur@31 286 return self
awgur@31 287
awgur@31 288 def __exit__(self, exc_type, exc_val, exc_tb):
awgur@31 289 raise NotImplemented()
awgur@31 290
awgur@31 291
awgur@31 292 class CPHAbsentSection(CPHSectionBase):
awgur@31 293 """\
awgur@31 294 Класс создаваемый на отсутствующую секцию конфигурационного файла
awgur@31 295 """
awgur@31 296 def get(self, config_prop_name: str, dc_prop_name: str):
awgur@31 297 raise NoSectionNotification()
awgur@31 298
awgur@31 299 def __exit__(self, exc_type, exc_val, exc_tb):
awgur@31 300 if exc_type == NoSectionNotification:
awgur@31 301 return True
awgur@31 302
awgur@31 303
awgur@31 304 class CPHParamGetter(CPHSectionBase):
awgur@31 305 def __init__(self, parser_helper_object):
awgur@31 306 self.ph = parser_helper_object
awgur@31 307 self.params = {}
awgur@31 308
awgur@34 309 def _add_param(self, param_name: str, param_val: Any, parser: Optional[Callable[[Any], Any]] = None):
awgur@31 310 """\
awgur@31 311 Непосредственное добавление полученного параметра со всеми проверками.
awgur@31 312 """
awgur@31 313
awgur@34 314 if parser is not None and param_val is not None:
awgur@34 315 param_val = parser(param_val)
awgur@34 316
awgur@31 317 fld = self.ph.fields.get(param_name)
awgur@31 318 if not isinstance(fld, Field):
awgur@31 319 raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '
awgur@31 320 f'которое мы должны заполнить из параметра конфигурации: {fld}')
awgur@31 321
awgur@31 322 if param_val is not None:
awgur@36 323 type_desc = get_type_describer(fld.type)
awgur@31 324 try:
awgur@36 325 res = type_desc(param_val)
awgur@31 326
awgur@31 327 except (ValueError, TypeError) as e:
awgur@36 328 raise ConfigParseHelperError(f'При приведении параметра к '
awgur@36 329 f'заданному типу произошла ошибка: '
awgur@36 330 f'значение="{param_val}" ошибка="{e}"')
awgur@31 331
awgur@31 332 else:
awgur@31 333 if fld.default is not MISSING:
awgur@31 334 res = fld.default
awgur@31 335
awgur@31 336 elif fld.default_factory is not MISSING:
awgur@31 337 res = fld.default_factory()
awgur@31 338
awgur@31 339 else:
awgur@31 340 raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')
awgur@31 341
awgur@31 342 self.params[param_name] = res
awgur@31 343
awgur@31 344 def __exit__(self, exc_type, exc_val, exc_tb):
awgur@31 345 if exc_type is None:
awgur@31 346 self.ph.add_params(self.params)
awgur@31 347
awgur@34 348 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
awgur@31 349 raise NoSectionNotification()
awgur@31 350
awgur@31 351
awgur@31 352 class CPHSection(CPHParamGetter):
awgur@31 353 """\
awgur@31 354 Класс производящий разбор конкретной секции конфигурации
awgur@31 355 """
awgur@31 356
awgur@31 357 def __init__(self, parser_helper_object, section: str):
awgur@31 358 super().__init__(parser_helper_object)
awgur@31 359 self.section_name = section
awgur@31 360 self.section = parser_helper_object.conf_parser[section]
awgur@31 361
awgur@34 362 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
awgur@31 363 """\
awgur@31 364 :param config_prop_name: Имя опции в файле конфигурации
awgur@31 365 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
awgur@34 366 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
awgur@31 367 """
awgur@31 368 try:
awgur@34 369 self._add_param(dc_prop_name, self.section.get(config_prop_name), parser)
awgur@31 370
awgur@31 371 except ConfigParseHelperError as e:
awgur@31 372 raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '
awgur@31 373 f'в секции "{self.section_name}": {e}')
awgur@31 374
awgur@31 375
awgur@31 376 class CPHEnvParser(CPHParamGetter):
awgur@31 377 """\
awgur@31 378 Класс для разбора переменных окружения в том же ключе, что и файла конфигурации
awgur@31 379 """
awgur@31 380
awgur@34 381 def get(self, env_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
awgur@31 382 """\
awgur@31 383 :param env_name: Имя переменной окружения, хранящей опцию
awgur@31 384 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
awgur@34 385 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
awgur@31 386 """
awgur@31 387
awgur@31 388 try:
awgur@34 389 self._add_param(dc_prop_name, getenv(env_name), parser)
awgur@31 390
awgur@31 391 except ConfigParseHelperError as e:
awgur@31 392 raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')
awgur@31 393
awgur@31 394
awgur@32 395 class CPHObjectsListGetter:
awgur@32 396 """\
awgur@32 397 Помощник для случаев, когда в наборе секций хранится однотипный набор объектов
awgur@32 398 """
awgur@32 399 def __init__(self, config_object_class: type, config_parser: ConfigParser, sections: Iterable[str]):
awgur@32 400 self.sections = sections
awgur@32 401 self.conf_parser = config_parser
awgur@32 402
awgur@32 403 if not is_dataclass(config_object_class):
awgur@32 404 raise ConfigParseHelperError(f'Представленный в качестве представления объекта конфигурации '
awgur@32 405 f'класс не является классом данных: {config_object_class.__name__}')
awgur@32 406
awgur@32 407 self.res_obj = config_object_class
awgur@32 408 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
awgur@32 409 self.obj_list = []
awgur@32 410 self.ident_list = []
awgur@32 411
awgur@32 412 def add_params(self, params: Dict[str, Any]):
awgur@32 413 try:
awgur@32 414 self.obj_list.append(self.res_obj(**params))
awgur@32 415
awgur@32 416 except (ValueError, TypeError) as e:
awgur@32 417 raise ConfigParseHelperError(f'Ошибка создания объекта объекта конфигурации, '
awgur@32 418 f'списка объектов конфигурации: {e}')
awgur@32 419
awgur@34 420 def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
awgur@32 421 """\
awgur@32 422 Подготавливаем список соответствия названий параметров в секции конкретным свойствам данного
awgur@32 423 в помощник класса
awgur@32 424
awgur@32 425 :param config_prop_name: Имя опции в файле конфигурации
awgur@32 426 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
awgur@34 427 :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
awgur@32 428 """
awgur@32 429
awgur@34 430 self.ident_list.append((config_prop_name, dc_prop_name, parser))
awgur@32 431
awgur@32 432 def get_config_objects(self) -> List[object]:
awgur@32 433 for section in self.sections:
awgur@32 434 try:
awgur@32 435 with CPHSection(self, section) as section_helper:
awgur@34 436 for conf_prop, dc_prop, parser in self.ident_list:
awgur@34 437 section_helper.get(conf_prop, dc_prop, parser)
awgur@32 438
awgur@32 439 except ConfigParseHelperError as e:
awgur@32 440 raise ConfigParseHelperError(f'Ошибка при разборе секции "{section}": {e}')
awgur@32 441
awgur@32 442 res = self.obj_list[:]
awgur@32 443 self.obj_list.clear()
awgur@32 444
awgur@32 445 return res
awgur@32 446
awgur@32 447
awgur@31 448 class ConfigParseHelper:
awgur@32 449 """\
awgur@32 450 Помощник разбора конфигурации
awgur@32 451 """
awgur@31 452 def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):
awgur@31 453 """\
awgur@31 454 :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации
awgur@31 455 :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле
awgur@31 456 """
awgur@31 457
awgur@31 458 if required_sections is not None:
awgur@31 459 self.req_sections = set(required_sections)
awgur@31 460
awgur@31 461 else:
awgur@31 462 self.req_sections = set()
awgur@31 463
awgur@31 464 if not is_dataclass(config_object_class):
awgur@31 465 raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '
awgur@31 466 f'классом данных: {config_object_class.__name__}')
awgur@31 467
awgur@31 468 self.res_obj = config_object_class
awgur@31 469 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
awgur@31 470 self.conf_parser: Optional[ConfigParser] = None
awgur@31 471 self.config_params = {}
awgur@31 472 self.config_params_lock = RLock()
awgur@31 473
awgur@31 474 def add_params(self, params: Dict[str, Any]):
awgur@31 475 self.config_params_lock.acquire()
awgur@31 476 try:
awgur@31 477 self.config_params.update(params)
awgur@31 478
awgur@31 479 finally:
awgur@31 480 self.config_params_lock.release()
awgur@31 481
awgur@31 482 def section(self, section_name: str) -> CPHSectionBase:
awgur@31 483 if self.conf_parser is None:
awgur@31 484 raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')
awgur@31 485
awgur@31 486 if self.conf_parser.has_section(section_name):
awgur@31 487 return CPHSection(self, section_name)
awgur@31 488
awgur@31 489 else:
awgur@31 490 return CPHAbsentSection()
awgur@31 491
awgur@31 492 def load(self, filename: str):
awgur@31 493 res = ConfigParser()
awgur@31 494 try:
awgur@31 495 res.read(filename)
awgur@31 496
awgur@31 497 except (TypeError, IOError, OSError, ValueError) as e:
awgur@31 498 raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '
awgur@31 499 f'ошибка="{e}"')
awgur@31 500
awgur@31 501 missing_sections = self.req_sections - set(res.sections())
awgur@31 502
awgur@31 503 if missing_sections:
awgur@31 504 missing_sections = ', '.join(missing_sections)
awgur@31 505 raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')
awgur@31 506
awgur@31 507 self.conf_parser = res
awgur@31 508
awgur@31 509 def get_config(self):
awgur@31 510 try:
awgur@31 511 return self.res_obj(**self.config_params)
awgur@31 512
awgur@31 513 except (ValueError, TypeError) as e:
awgur@31 514 raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')
awgur@32 515
awgur@32 516 def get_sections(self):
awgur@32 517 return self.conf_parser.sections()