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