py.lib

     1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/type_utils/config_parse_helper.py	Sat Aug 20 23:56:16 2022 +0300
     1.3 @@ -0,0 +1,272 @@
     1.4 +# coding: utf-8
     1.5 +"""\
     1.6 +Получение из конфигурационного файла параметров в виде класса данных.
     1.7 +
     1.8 +НЕ РАБОТАЕТ БЕЗ ОСТАЛЬНОГО МОДУЛЯ type_utils
     1.9 +"""
    1.10 +
    1.11 +from configparser import ConfigParser
    1.12 +from dataclasses import is_dataclass, fields, Field, MISSING
    1.13 +from typing import Iterable, Optional, Dict, Any, List, Callable, Union
    1.14 +from threading import RLock
    1.15 +from os import getenv
    1.16 +
    1.17 +from .type_descriptor import get_type_describer
    1.18 +
    1.19 +
    1.20 +class ConfigParseHelperError(Exception):
    1.21 +    """\
    1.22 +    Ошибки в модуле помощника разборщика конфигураций
    1.23 +    """
    1.24 +
    1.25 +
    1.26 +class NoSectionNotification(ConfigParseHelperError):
    1.27 +    """\
    1.28 +    Оповещение об отсутствующей секции конфигурации
    1.29 +    """
    1.30 +
    1.31 +
    1.32 +class CPHSectionBase:
    1.33 +    """\
    1.34 +    Базовый класс обработки секции конфигурационного файла
    1.35 +    """
    1.36 +
    1.37 +    def get(self, config_prop_name: str, dc_prop_name: str):
    1.38 +        """\
    1.39 +        Получить свойство из конфигурационного файла
    1.40 +        """
    1.41 +        raise NotImplemented()
    1.42 +
    1.43 +    def __enter__(self):
    1.44 +        return self
    1.45 +
    1.46 +    def __exit__(self, exc_type, exc_val, exc_tb):
    1.47 +        raise NotImplemented()
    1.48 +
    1.49 +
    1.50 +class CPHAbsentSection(CPHSectionBase):
    1.51 +    """\
    1.52 +    Класс создаваемый на отсутствующую секцию конфигурационного файла
    1.53 +    """
    1.54 +    def get(self, config_prop_name: str, dc_prop_name: str):
    1.55 +        raise NoSectionNotification()
    1.56 +
    1.57 +    def __exit__(self, exc_type, exc_val, exc_tb):
    1.58 +        if exc_type == NoSectionNotification:
    1.59 +            return True
    1.60 +
    1.61 +
    1.62 +class CPHParamGetter(CPHSectionBase):
    1.63 +    def __init__(self, parser_helper_object):
    1.64 +        self.ph = parser_helper_object
    1.65 +        self.params = {}
    1.66 +
    1.67 +    def _add_param(self, param_name: str, param_val: Any, parser: Optional[Callable[[Any], Any]] = None):
    1.68 +        """\
    1.69 +        Непосредственное добавление полученного параметра со всеми проверками.
    1.70 +        """
    1.71 +
    1.72 +        if parser is not None and param_val is not None:
    1.73 +            param_val = parser(param_val)
    1.74 +
    1.75 +        fld = self.ph.fields.get(param_name)
    1.76 +        if not isinstance(fld, Field):
    1.77 +            raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '
    1.78 +                                         f'которое мы должны заполнить из параметра конфигурации: {fld}')
    1.79 +
    1.80 +        if param_val is not None:
    1.81 +            type_desc = get_type_describer(fld.type)
    1.82 +            try:
    1.83 +                res = type_desc(param_val)
    1.84 +
    1.85 +            except (ValueError, TypeError) as e:
    1.86 +                raise ConfigParseHelperError(f'При приведении параметра к '
    1.87 +                                             f'заданному типу произошла ошибка: '
    1.88 +                                             f'значение="{param_val}" ошибка="{e}"')
    1.89 +
    1.90 +        else:
    1.91 +            if fld.default is not MISSING:
    1.92 +                res = fld.default
    1.93 +
    1.94 +            elif fld.default_factory is not MISSING:
    1.95 +                res = fld.default_factory()
    1.96 +
    1.97 +            else:
    1.98 +                raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')
    1.99 +
   1.100 +        self.params[param_name] = res
   1.101 +
   1.102 +    def __exit__(self, exc_type, exc_val, exc_tb):
   1.103 +        if exc_type is None:
   1.104 +            self.ph.add_params(self.params)
   1.105 +
   1.106 +    def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
   1.107 +        raise NoSectionNotification()
   1.108 +
   1.109 +
   1.110 +class CPHSection(CPHParamGetter):
   1.111 +    """\
   1.112 +    Класс производящий разбор конкретной секции конфигурации
   1.113 +    """
   1.114 +
   1.115 +    def __init__(self, parser_helper_object, section: str):
   1.116 +        super().__init__(parser_helper_object)
   1.117 +        self.section_name = section
   1.118 +        self.section = parser_helper_object.conf_parser[section]
   1.119 +
   1.120 +    def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
   1.121 +        """\
   1.122 +        :param config_prop_name: Имя опции в файле конфигурации
   1.123 +        :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
   1.124 +        :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
   1.125 +        """
   1.126 +        try:
   1.127 +            self._add_param(dc_prop_name, self.section.get(config_prop_name), parser)
   1.128 +
   1.129 +        except ConfigParseHelperError as e:
   1.130 +            raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '
   1.131 +                                         f'в секции "{self.section_name}": {e}')
   1.132 +
   1.133 +
   1.134 +class CPHEnvParser(CPHParamGetter):
   1.135 +    """\
   1.136 +    Класс для разбора переменных окружения в том же ключе, что и файла конфигурации
   1.137 +    """
   1.138 +
   1.139 +    def get(self, env_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
   1.140 +        """\
   1.141 +        :param env_name: Имя переменной окружения, хранящей опцию
   1.142 +        :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
   1.143 +        :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
   1.144 +        """
   1.145 +
   1.146 +        try:
   1.147 +            self._add_param(dc_prop_name, getenv(env_name), parser)
   1.148 +
   1.149 +        except ConfigParseHelperError as e:
   1.150 +            raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')
   1.151 +
   1.152 +
   1.153 +class CPHObjectsListGetter:
   1.154 +    """\
   1.155 +    Помощник для случаев, когда в наборе секций хранится однотипный набор объектов
   1.156 +    """
   1.157 +    def __init__(self, config_object_class: type, config_parser: ConfigParser, sections: Iterable[str]):
   1.158 +        self.sections = sections
   1.159 +        self.conf_parser = config_parser
   1.160 +
   1.161 +        if not is_dataclass(config_object_class):
   1.162 +            raise ConfigParseHelperError(f'Представленный в качестве представления объекта конфигурации '
   1.163 +                                         f'класс не является классом данных: {config_object_class.__name__}')
   1.164 +
   1.165 +        self.res_obj = config_object_class
   1.166 +        self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
   1.167 +        self.obj_list = []
   1.168 +        self.ident_list = []
   1.169 +
   1.170 +    def add_params(self, params: Dict[str, Any]):
   1.171 +        try:
   1.172 +            self.obj_list.append(self.res_obj(**params))
   1.173 +
   1.174 +        except (ValueError, TypeError) as e:
   1.175 +            raise ConfigParseHelperError(f'Ошибка создания объекта объекта конфигурации, '
   1.176 +                                         f'списка объектов конфигурации: {e}')
   1.177 +
   1.178 +    def get(self, config_prop_name: str, dc_prop_name: str, parser: Optional[Callable[[Any], Any]] = None):
   1.179 +        """\
   1.180 +        Подготавливаем список соответствия названий параметров в секции конкретным свойствам данного
   1.181 +        в помощник класса
   1.182 +
   1.183 +        :param config_prop_name: Имя опции в файле конфигурации
   1.184 +        :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
   1.185 +        :param parser: Исполнимый обработчик значения, перед его помещением в конфигурацию
   1.186 +        """
   1.187 +
   1.188 +        self.ident_list.append((config_prop_name, dc_prop_name, parser))
   1.189 +
   1.190 +    def get_config_objects(self) -> List[object]:
   1.191 +        for section in self.sections:
   1.192 +            try:
   1.193 +                with CPHSection(self, section) as section_helper:
   1.194 +                    for conf_prop, dc_prop, parser in self.ident_list:
   1.195 +                        section_helper.get(conf_prop, dc_prop, parser)
   1.196 +
   1.197 +            except ConfigParseHelperError as e:
   1.198 +                raise ConfigParseHelperError(f'Ошибка при разборе секции "{section}": {e}')
   1.199 +
   1.200 +        res = self.obj_list[:]
   1.201 +        self.obj_list.clear()
   1.202 +
   1.203 +        return res
   1.204 +
   1.205 +
   1.206 +class ConfigParseHelper:
   1.207 +    """\
   1.208 +    Помощник разбора конфигурации
   1.209 +    """
   1.210 +    def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):
   1.211 +        """\
   1.212 +        :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации
   1.213 +        :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле
   1.214 +        """
   1.215 +
   1.216 +        if required_sections is not None:
   1.217 +            self.req_sections = set(required_sections)
   1.218 +
   1.219 +        else:
   1.220 +            self.req_sections = set()
   1.221 +
   1.222 +        if not is_dataclass(config_object_class):
   1.223 +            raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '
   1.224 +                                         f'классом данных: {config_object_class.__name__}')
   1.225 +
   1.226 +        self.res_obj = config_object_class
   1.227 +        self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
   1.228 +        self.conf_parser: Optional[ConfigParser] = None
   1.229 +        self.config_params = {}
   1.230 +        self.config_params_lock = RLock()
   1.231 +
   1.232 +    def add_params(self, params: Dict[str, Any]):
   1.233 +        self.config_params_lock.acquire()
   1.234 +        try:
   1.235 +            self.config_params.update(params)
   1.236 +
   1.237 +        finally:
   1.238 +            self.config_params_lock.release()
   1.239 +
   1.240 +    def section(self, section_name: str) -> CPHSectionBase:
   1.241 +        if self.conf_parser is None:
   1.242 +            raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')
   1.243 +
   1.244 +        if self.conf_parser.has_section(section_name):
   1.245 +            return CPHSection(self, section_name)
   1.246 +
   1.247 +        else:
   1.248 +            return CPHAbsentSection()
   1.249 +
   1.250 +    def load(self, filename: str):
   1.251 +        res = ConfigParser()
   1.252 +        try:
   1.253 +            res.read(filename)
   1.254 +
   1.255 +        except (TypeError, IOError, OSError, ValueError) as e:
   1.256 +            raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '
   1.257 +                                         f'ошибка="{e}"')
   1.258 +
   1.259 +        missing_sections = self.req_sections - set(res.sections())
   1.260 +
   1.261 +        if missing_sections:
   1.262 +            missing_sections = ', '.join(missing_sections)
   1.263 +            raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')
   1.264 +
   1.265 +        self.conf_parser = res
   1.266 +
   1.267 +    def get_config(self):
   1.268 +        try:
   1.269 +            return self.res_obj(**self.config_params)
   1.270 +
   1.271 +        except (ValueError, TypeError) as e:
   1.272 +            raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')
   1.273 +
   1.274 +    def get_sections(self):
   1.275 +        return self.conf_parser.sections()