py.lib

Yohn Y. 2022-08-13 Parent:4186c3b229fa Child:57f63bf31fd8

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