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