py.lib

     1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/config_parse_helper.py	Fri Aug 05 23:58:12 2022 +0300
     1.3 @@ -0,0 +1,199 @@
     1.4 +# coding: utf-8
     1.5 +from configparser import ConfigParser
     1.6 +from dataclasses import is_dataclass, fields, Field, MISSING
     1.7 +from typing import Iterable, Optional, Dict, Any
     1.8 +from threading import RLock
     1.9 +from os import getenv
    1.10 +
    1.11 +
    1.12 +class ConfigParseHelperError(Exception):
    1.13 +    """\
    1.14 +    Ошибки в модуле помощника разборщика конфигураций
    1.15 +    """
    1.16 +
    1.17 +
    1.18 +class NoSectionNotification(ConfigParseHelperError):
    1.19 +    """\
    1.20 +    Оповещение об отсутствующей секции конфигурации
    1.21 +    """
    1.22 +
    1.23 +
    1.24 +class CPHSectionBase:
    1.25 +    """\
    1.26 +    Базовый класс обработки секции конфигурационного файла
    1.27 +    """
    1.28 +
    1.29 +    def get(self, config_prop_name: str, dc_prop_name: str):
    1.30 +        """\
    1.31 +        Получить свойство из конфигурационного файла
    1.32 +        """
    1.33 +        raise NotImplemented()
    1.34 +
    1.35 +    def __enter__(self):
    1.36 +        return self
    1.37 +
    1.38 +    def __exit__(self, exc_type, exc_val, exc_tb):
    1.39 +        raise NotImplemented()
    1.40 +
    1.41 +
    1.42 +class CPHAbsentSection(CPHSectionBase):
    1.43 +    """\
    1.44 +    Класс создаваемый на отсутствующую секцию конфигурационного файла
    1.45 +    """
    1.46 +    def get(self, config_prop_name: str, dc_prop_name: str):
    1.47 +        raise NoSectionNotification()
    1.48 +
    1.49 +    def __exit__(self, exc_type, exc_val, exc_tb):
    1.50 +        if exc_type == NoSectionNotification:
    1.51 +            return True
    1.52 +
    1.53 +
    1.54 +class CPHParamGetter(CPHSectionBase):
    1.55 +    def __init__(self, parser_helper_object):
    1.56 +        self.ph = parser_helper_object
    1.57 +        self.params = {}
    1.58 +
    1.59 +    def _add_param(self, param_name: str, param_val: Any):
    1.60 +        """\
    1.61 +        Непосредственное добавление полученного параметра со всеми проверками.
    1.62 +        """
    1.63 +
    1.64 +        fld = self.ph.fields.get(param_name)
    1.65 +        if not isinstance(fld, Field):
    1.66 +            raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '
    1.67 +                                         f'которое мы должны заполнить из параметра конфигурации: {fld}')
    1.68 +
    1.69 +        if param_val is not None:
    1.70 +            try:
    1.71 +                res = fld.type(param_val)
    1.72 +
    1.73 +            except (ValueError, TypeError) as e:
    1.74 +                raise ConfigParseHelperError(f'При приведении параметра к '
    1.75 +                                             f'заданному типу произошла ошибка: '
    1.76 +                                             f'значение="{param_val}" ошибка="{e}"')
    1.77 +
    1.78 +        else:
    1.79 +            if fld.default is not MISSING:
    1.80 +                res = fld.default
    1.81 +
    1.82 +            elif fld.default_factory is not MISSING:
    1.83 +                res = fld.default_factory()
    1.84 +
    1.85 +            else:
    1.86 +                raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')
    1.87 +
    1.88 +        self.params[param_name] = res
    1.89 +
    1.90 +    def __exit__(self, exc_type, exc_val, exc_tb):
    1.91 +        if exc_type is None:
    1.92 +            self.ph.add_params(self.params)
    1.93 +
    1.94 +    def get(self, config_prop_name: str, dc_prop_name: str):
    1.95 +        raise NoSectionNotification()
    1.96 +
    1.97 +
    1.98 +class CPHSection(CPHParamGetter):
    1.99 +    """\
   1.100 +    Класс производящий разбор конкретной секции конфигурации
   1.101 +    """
   1.102 +
   1.103 +    def __init__(self, parser_helper_object, section: str):
   1.104 +        super().__init__(parser_helper_object)
   1.105 +        self.section_name = section
   1.106 +        self.section = parser_helper_object.conf_parser[section]
   1.107 +
   1.108 +    def get(self, config_prop_name: str, dc_prop_name: str):
   1.109 +        """\
   1.110 +        :param config_prop_name: Имя опции в файле конфигурации
   1.111 +        :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
   1.112 +        """
   1.113 +        try:
   1.114 +            self._add_param(dc_prop_name, self.section.get(config_prop_name))
   1.115 +
   1.116 +        except ConfigParseHelperError as e:
   1.117 +            raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '
   1.118 +                                         f'в секции "{self.section_name}": {e}')
   1.119 +
   1.120 +
   1.121 +class CPHEnvParser(CPHParamGetter):
   1.122 +    """\
   1.123 +    Класс для разбора переменных окружения в том же ключе, что и файла конфигурации
   1.124 +    """
   1.125 +
   1.126 +    def get(self, env_name: str, dc_prop_name: str):
   1.127 +        """\
   1.128 +        :param env_name: Имя переменной окружения, хранящей опцию
   1.129 +        :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию
   1.130 +        """
   1.131 +
   1.132 +        try:
   1.133 +            self._add_param(dc_prop_name, getenv(env_name))
   1.134 +
   1.135 +        except ConfigParseHelperError as e:
   1.136 +            raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')
   1.137 +
   1.138 +
   1.139 +class ConfigParseHelper:
   1.140 +    def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):
   1.141 +        """\
   1.142 +        :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации
   1.143 +        :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле
   1.144 +        """
   1.145 +
   1.146 +        if required_sections is not None:
   1.147 +            self.req_sections = set(required_sections)
   1.148 +
   1.149 +        else:
   1.150 +            self.req_sections = set()
   1.151 +
   1.152 +        if not is_dataclass(config_object_class):
   1.153 +            raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '
   1.154 +                                         f'классом данных: {config_object_class.__name__}')
   1.155 +
   1.156 +        self.res_obj = config_object_class
   1.157 +        self.fields = dict((fld.name, fld) for fld in fields(config_object_class))
   1.158 +        self.conf_parser: Optional[ConfigParser] = None
   1.159 +        self.config_params = {}
   1.160 +        self.config_params_lock = RLock()
   1.161 +
   1.162 +    def add_params(self, params: Dict[str, Any]):
   1.163 +        self.config_params_lock.acquire()
   1.164 +        try:
   1.165 +            self.config_params.update(params)
   1.166 +
   1.167 +        finally:
   1.168 +            self.config_params_lock.release()
   1.169 +
   1.170 +    def section(self, section_name: str) -> CPHSectionBase:
   1.171 +        if self.conf_parser is None:
   1.172 +            raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')
   1.173 +
   1.174 +        if self.conf_parser.has_section(section_name):
   1.175 +            return CPHSection(self, section_name)
   1.176 +
   1.177 +        else:
   1.178 +            return CPHAbsentSection()
   1.179 +
   1.180 +    def load(self, filename: str):
   1.181 +        res = ConfigParser()
   1.182 +        try:
   1.183 +            res.read(filename)
   1.184 +
   1.185 +        except (TypeError, IOError, OSError, ValueError) as e:
   1.186 +            raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '
   1.187 +                                         f'ошибка="{e}"')
   1.188 +
   1.189 +        missing_sections = self.req_sections - set(res.sections())
   1.190 +
   1.191 +        if missing_sections:
   1.192 +            missing_sections = ', '.join(missing_sections)
   1.193 +            raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')
   1.194 +
   1.195 +        self.conf_parser = res
   1.196 +
   1.197 +    def get_config(self):
   1.198 +        try:
   1.199 +            return self.res_obj(**self.config_params)
   1.200 +
   1.201 +        except (ValueError, TypeError) as e:
   1.202 +            raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')