py.lib

Yohn Y. 2022-08-13 Parent:eb41cc498ddb Child:84b54a8a6d4c

py.lib/config_parse_helper.py

* С разбегу влетел в type hinting, пришлось обкладывать костылями. И этот процесс судя по всему длительный, и место больное будет долго...

Download raw file

View source Diff to previous Annotate

1 # coding: utf-8

2 from configparser import ConfigParser

3 from dataclasses import is_dataclass, fields, Field, MISSING

4 from typing import Iterable, Optional, Dict, Any, List

5 from threading import RLock

6 from os import getenv

9 class ConfigParseHelperError(Exception):

10 """\

11 Ошибки в модуле помощника разборщика конфигураций

12 """

15 class NoSectionNotification(ConfigParseHelperError):

16 """\

17 Оповещение об отсутствующей секции конфигурации

18 """

21 def difficult_type_recognizer(type_obj, value):

22 """\

23 Магия борющаяся с костылями type hinting

24 """

25 if type(type_obj).__name__ == '_GenericAlias':

26 if (

27 hasattr(type_obj, '__args__')

28 and type(type_obj.__args__) == tuple

29 and type_obj.__args__

30 ):

31 if type_obj.__name__ == 'List':

32 return list(map(type_obj.__args__[0], value))

34 elif type_obj.__name__ == 'Tuple':

35 return tuple(map(type_obj.__args__[0], value))

37 else:

38 raise ValueError('Неизвестный тип')

40 else:

41 ValueError('Неизвестный тип')

43 elif type(type_obj).__name__ == '_UnionGenericAlias':

44 if type_obj.__name__ not in ('Union', 'Optional'):

45 raise TypeError(f'Неизвестный подтип _UnionGenericAlias: {type_obj.__name__}')

47 if not (

48 hasattr(type_obj, '__args__')

49 and type(type_obj.__args__) == tuple

50 and type_obj.__args__

51 ):

52 raise TypeError(f'Не ясно как работать с типом не вижу аргументов: {type_obj.__name__}')

54 for _t in type_obj.__args__:

55 if _t.__name__ == 'NoneType':

56 continue

58 try:

59 return _t(value)

61 except (TypeError, ValueError):

62 continue

64 raise ValueError('Не удалось привести значение к типу')

67 class CPHSectionBase:

68 """\

69 Базовый класс обработки секции конфигурационного файла

70 """

72 def get(self, config_prop_name: str, dc_prop_name: str):

73 """\

74 Получить свойство из конфигурационного файла

75 """

76 raise NotImplemented()

78 def __enter__(self):

79 return self

81 def __exit__(self, exc_type, exc_val, exc_tb):

82 raise NotImplemented()

85 class CPHAbsentSection(CPHSectionBase):

86 """\

87 Класс создаваемый на отсутствующую секцию конфигурационного файла

88 """

89 def get(self, config_prop_name: str, dc_prop_name: str):

90 raise NoSectionNotification()

92 def __exit__(self, exc_type, exc_val, exc_tb):

93 if exc_type == NoSectionNotification:

94 return True

97 class CPHParamGetter(CPHSectionBase):

98 def __init__(self, parser_helper_object):

99 self.ph = parser_helper_object

100 self.params = {}

101

102 def _add_param(self, param_name: str, param_val: Any):

103 """\

104 Непосредственное добавление полученного параметра со всеми проверками.

105 """

106

107 fld = self.ph.fields.get(param_name)

108 if not isinstance(fld, Field):

109 raise ConfigParseHelperError(f'В классе данных отсутствует свойство "{param_name}", '

110 f'которое мы должны заполнить из параметра конфигурации: {fld}')

111

112 if param_val is not None:

113 try:

114 res = fld.type(param_val)

115

116 except (ValueError, TypeError) as e:

117 try:

118 res = difficult_type_recognizer(fld.type, param_val)

119

120 except (ValueError, TypeError):

121 raise ConfigParseHelperError(f'При приведении параметра к '

122 f'заданному типу произошла ошибка: '

123 f'значение="{param_val}" ошибка="{e}"')

124

125 else:

126 if fld.default is not MISSING:

127 res = fld.default

128

129 elif fld.default_factory is not MISSING:

130 res = fld.default_factory()

131

132 else:

133 raise ConfigParseHelperError('В конфигурации не заданна обязательная опция')

134

135 self.params[param_name] = res

136

137 def __exit__(self, exc_type, exc_val, exc_tb):

138 if exc_type is None:

139 self.ph.add_params(self.params)

140

141 def get(self, config_prop_name: str, dc_prop_name: str):

142 raise NoSectionNotification()

143

144

145 class CPHSection(CPHParamGetter):

146 """\

147 Класс производящий разбор конкретной секции конфигурации

148 """

149

150 def __init__(self, parser_helper_object, section: str):

151 super().__init__(parser_helper_object)

152 self.section_name = section

153 self.section = parser_helper_object.conf_parser[section]

154

155 def get(self, config_prop_name: str, dc_prop_name: str):

156 """\

157 :param config_prop_name: Имя опции в файле конфигурации

158 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию

159 """

160 try:

161 self._add_param(dc_prop_name, self.section.get(config_prop_name))

162

163 except ConfigParseHelperError as e:

164 raise ConfigParseHelperError(f'Ошибка при разборе параметра "{config_prop_name}" '

165 f'в секции "{self.section_name}": {e}')

166

167

168 class CPHEnvParser(CPHParamGetter):

169 """\

170 Класс для разбора переменных окружения в том же ключе, что и файла конфигурации

171 """

172

173 def get(self, env_name: str, dc_prop_name: str):

174 """\

175 :param env_name: Имя переменной окружения, хранящей опцию

176 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию

177 """

178

179 try:

180 self._add_param(dc_prop_name, getenv(env_name))

181

182 except ConfigParseHelperError as e:

183 raise ConfigParseHelperError(f'Ошибка при получении значения из переменной окружения "{env_name}": {e}')

184

185

186 class CPHObjectsListGetter:

187 """\

188 Помощник для случаев, когда в наборе секций хранится однотипный набор объектов

189 """

190 def __init__(self, config_object_class: type, config_parser: ConfigParser, sections: Iterable[str]):

191 self.sections = sections

192 self.conf_parser = config_parser

193

194 if not is_dataclass(config_object_class):

195 raise ConfigParseHelperError(f'Представленный в качестве представления объекта конфигурации '

196 f'класс не является классом данных: {config_object_class.__name__}')

197

198 self.res_obj = config_object_class

199 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))

200 self.obj_list = []

201 self.ident_list = []

202

203 def add_params(self, params: Dict[str, Any]):

204 try:

205 self.obj_list.append(self.res_obj(**params))

206

207 except (ValueError, TypeError) as e:

208 raise ConfigParseHelperError(f'Ошибка создания объекта объекта конфигурации, '

209 f'списка объектов конфигурации: {e}')

210

211 def get(self, config_prop_name: str, dc_prop_name: str):

212 """\

213 Подготавливаем список соответствия названий параметров в секции конкретным свойствам данного

214 в помощник класса

215

216 :param config_prop_name: Имя опции в файле конфигурации

217 :param dc_prop_name: Имя свойства в классе данных, хранящем конфигурацию

218 """

219

220 self.ident_list.append((config_prop_name, dc_prop_name))

221

222 def get_config_objects(self) -> List[object]:

223 for section in self.sections:

224 try:

225 with CPHSection(self, section) as section_helper:

226 for conf_prop, dc_prop in self.ident_list:

227 section_helper.get(conf_prop, dc_prop)

228

229 except ConfigParseHelperError as e:

230 raise ConfigParseHelperError(f'Ошибка при разборе секции "{section}": {e}')

231

232 res = self.obj_list[:]

233 self.obj_list.clear()

234

235 return res

236

237

238 class ConfigParseHelper:

239 """\

240 Помощник разбора конфигурации

241 """

242 def __init__(self, config_object_class: type, required_sections: Optional[Iterable[str]] = None):

243 """\

244 :param config_object_class: Dataclass, который мы подготовили как хранилище параметров конфигурации

245 :param required_sections: Перечисление секций конфигурации, которые мы требуем, чтобы были в файле

246 """

247

248 if required_sections is not None:

249 self.req_sections = set(required_sections)

250

251 else:

252 self.req_sections = set()

253

254 if not is_dataclass(config_object_class):

255 raise ConfigParseHelperError(f'Представленный в качестве объекта конфигурации класс не является '

256 f'классом данных: {config_object_class.__name__}')

257

258 self.res_obj = config_object_class

259 self.fields = dict((fld.name, fld) for fld in fields(config_object_class))

260 self.conf_parser: Optional[ConfigParser] = None

261 self.config_params = {}

262 self.config_params_lock = RLock()

263

264 def add_params(self, params: Dict[str, Any]):

265 self.config_params_lock.acquire()

266 try:

267 self.config_params.update(params)

268

269 finally:

270 self.config_params_lock.release()

271

272 def section(self, section_name: str) -> CPHSectionBase:

273 if self.conf_parser is None:

274 raise ConfigParseHelperError(f'Прежде чем приступать к разбору файла конфигурации стоит его загрузить')

275

276 if self.conf_parser.has_section(section_name):

277 return CPHSection(self, section_name)

278

279 else:

280 return CPHAbsentSection()

281

282 def load(self, filename: str):

283 res = ConfigParser()

284 try:

285 res.read(filename)

286

287 except (TypeError, IOError, OSError, ValueError) as e:

288 raise ConfigParseHelperError(f'Не удалось загрузить файл конфигурации: файл="{filename}" '

289 f'ошибка="{e}"')

290

291 missing_sections = self.req_sections - set(res.sections())

292

293 if missing_sections:

294 missing_sections = ', '.join(missing_sections)

295 raise ConfigParseHelperError(f'В конфигурационном файле отсутствуют секции: {missing_sections}')

296

297 self.conf_parser = res

298

299 def get_config(self):

300 try:

301 return self.res_obj(**self.config_params)

302

303 except (ValueError, TypeError) as e:

304 raise ConfigParseHelperError(f'Не удалось инициализировать объект конфигурации: {e}')

305

306 def get_sections(self):

307 return self.conf_parser.sections()