py.lib

Yohn Y. 2022-08-19 Parent:84b54a8a6d4c Child:ae0107755941

36:f1a05e880961 Go to Latest

py.lib/config_parse_helper.py

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