Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/confi/confi.py: 80%

1"""Easy Python configuration interface built on top of python-decouple and 

2click / typer. 

3 

4Adding typing support and parsing with Pydantic and Enum. 

5""" 

6 

7import inspect 

8import json 

9import logging 

10import string 

11from collections import OrderedDict 

12from functools import partial, wraps 

13from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union 

14 

15from decouple import Csv, UndefinedValueError, config, text_type, undefined 

16from opal_common.authentication.casting import cast_private_key, cast_public_key 

17from opal_common.authentication.types import EncryptionKeyFormat, PrivateKey, PublicKey 

18from opal_common.confi.cli import get_cli_object_for_config_objects 

19from opal_common.confi.types import ConfiDelay, ConfiEntry, no_cast 

20from opal_common.logging_utils.decorators import log_exception 

21from pydantic import BaseModel, ValidationError 

22from typer import Typer 

23 

24 

25class Placeholder(object): 

26 """Placeholder instead of default value for decouple.""" 

27 

28 pass 

29 

30 

31def cast_boolean(value): 

32 """Parse an entry as a boolean. 

33 

34 - all variations of "true" and 1 are treated as True 

35 - all variations of "false" and 0 are treated as False 

36 """ 

37 if isinstance(value, bool): 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true

38 return value 

39 elif isinstance(value, str): 39 ↛ 48line 39 didn't jump to line 48 because the condition on line 39 was always true

40 value = value.lower() 

41 if value == "true" or value == "1": 41 ↛ 43line 41 didn't jump to line 43 because the condition on line 41 was always true

42 return True 

43 elif value == "false" or value == "0": 

44 return False 

45 else: 

46 raise UndefinedValueError(f"{value} - is not a valid boolean") 

47 else: 

48 raise UndefinedValueError(f"{value} - is not a valid boolean") 

49 

50 

51def cast_pydantic(model: BaseModel): 

52 def cast_pydantic_by_model(value): 

53 if isinstance(value, str): 

54 return model.parse_raw(value) 

55 else: 

56 return model.parse_obj(value) 

57 

58 return cast_pydantic_by_model 

59 

60 

61def ignore_confi_delay_cast(cast_func): 

62 """When we pass a ConfiDelay as the default to decouple, until this delayed 

63 default is evaluated by confi, there is no point in casting it. 

64 

65 After a ConfiDelay is evaluated, the resulted value should be passed 

66 again to the cast method, and this time it will indeed be cast. 

67 """ 

68 

69 @wraps(cast_func) 

70 def wrapped_cast(value, *args, **kwargs): 

71 if isinstance(value, ConfiDelay): 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 return value 

73 return cast_func(value, *args, **kwargs) 

74 

75 return wrapped_cast 

76 

77 

78def load_conf_if_none(variable, conf): 

79 if variable is None: 

80 return conf 

81 else: 

82 return variable 

83 

84 

85EnumT = TypeVar("EnumT") 

86T = TypeVar("T", bound=BaseModel) 

87ValueT = TypeVar("ValueT") 

88 

89 

90class Confi: 

91 """Interface to create typed configuration entries.""" 

92 

93 def __init__(self, prefix=None, is_model=True) -> None: 

94 """ 

95 

96 Args: 

97 prefix (str, optional): Prefix to add to all env-var keys. Defaults to self.ENV_PREFIX (which defaults to ""). 

98 is_model (bool, optional): Should Confi.<type> return a ConfiEntry (the default, True) or should it evaluate env settings immediately and return a value (False) 

99 """ 

100 self._is_model = is_model 

101 self._prefix = prefix 

102 # counter of created entries (to track order) 

103 self._counter = 0 

104 # entries to be evaluated 

105 self._entries: Dict[str, ConfiEntry] = OrderedDict() 

106 # delayed entries to be evaluated (instead of being referenced by self._entries) 

107 self._delayed_entries: Dict[str, ConfiDelay] = OrderedDict() 

108 # entries with delayed defaults (in addition to being referenced by self._entries) 

109 self._delayed_defaults: Dict[str, ConfiEntry] = OrderedDict() 

110 

111 # get members by creation order 

112 members = sorted( 

113 inspect.getmembers(self, self._is_entry), key=self._get_entry_index 

114 ) 

115 # eval class entries into values (by order of definition - same order as in the config class lines) 

116 for name, entry in members: 

117 # unwrap delayed entries 

118 if isinstance(entry, ConfiDelay): 

119 entry = entry.eval(self) 

120 

121 if isinstance(entry, ConfiEntry): 121 ↛ 116line 121 didn't jump to line 116 because the condition on line 121 was always true

122 self._entries[name] = entry 

123 # save delayed 

124 if isinstance(entry.default, ConfiDelay): 

125 self._delayed_defaults[name] = entry 

126 # eval, and save the value into the class instance 

127 value = self._eval_and_save_entry(name, entry) 

128 # save the value into the entry to be used as default for CLI 

129 entry.value = value 

130 

131 # load (all calls inside should produce a real value) 

132 self._is_model = False 

133 

134 # load delayed values: 

135 for name, entry in self._delayed_defaults.items(): 

136 default: ConfiDelay = entry.default 

137 # but only if no value is set yet 

138 if entry.value == default or entry.value == undefined: 

139 setattr(self, name, entry.cast(default.eval(self))) 

140 

141 self.on_load() 

142 self._is_model = is_model 

143 

144 def _is_entry(self, entry): 

145 res = isinstance(entry, (ConfiEntry, ConfiDelay)) 

146 return res 

147 

148 def _get_entry_index(self, member: Tuple[str, ConfiEntry]): 

149 name, entry = member 

150 return entry.index 

151 

152 @property 

153 def entries(self): 

154 return self._entries 

155 

156 def _prefix_key(self, key): 

157 prefix = self._prefix 

158 return f"{prefix}{key}" if prefix is not None else key 

159 

160 def _eval_and_save_entry(self, name: str, entry: ConfiEntry): 

161 value = self._eval_entry(entry) 

162 setattr(self, name, value) 

163 return value 

164 

165 def _eval_entry(self, entry: ConfiEntry): 

166 whole_key = self._prefix_key(entry.key) 

167 res = self._evaluate(whole_key, entry.default, entry.cast, **entry.kwargs) 

168 return res 

169 

170 def _process( 

171 self, 

172 key, 

173 *, 

174 default=undefined, 

175 description=None, 

176 cast=no_cast, 

177 cast_from_json=no_cast, 

178 type: ValueT = str, 

179 flags: List[str] = None, 

180 **kwargs, 

181 ) -> Union[ValueT, ConfiEntry]: 

182 if self._is_model: 182 ↛ 199line 182 didn't jump to line 199 because the condition on line 182 was always true

183 # create new entry 

184 res = ConfiEntry( 

185 key, 

186 default=default, 

187 description=description, 

188 cast=cast, 

189 cast_from_json=cast_from_json, 

190 type=type, 

191 index=self._counter, 

192 flags=flags, 

193 **kwargs, 

194 ) 

195 # track count for indexing 

196 self._counter += 1 

197 return res 

198 

199 whole_key = self._prefix_key(key) 

200 return self._evaluate(whole_key, default, cast, **kwargs) 

201 

202 def _evaluate(self, key, default=undefined, cast=no_cast, **kwargs): 

203 safe_cast_func = ignore_confi_delay_cast(cast) 

204 # decouple expects a string don't pass actual objects to it, as it will try and cast them - instead pass undefined 

205 passed_default = default if isinstance(default, str) else undefined 

206 try: 

207 res = config(key, default=passed_default, cast=safe_cast_func, **kwargs) 

208 except UndefinedValueError: 

209 # return actual default if provided, if we don't have one re-raise 

210 if not isinstance(default, undefined.__class__): 210 ↛ 220line 210 didn't jump to line 220 because the condition on line 210 was always true

211 # cast the default value if needed (it's a string or a dict that represents an object); otherwise use as is 

212 if isinstance(default, str) or ( 

213 safe_cast_func.__name__ == cast_pydantic(BaseModel).__name__ 

214 and isinstance(default, dict) 

215 ): 

216 res = safe_cast_func(default) 

217 else: 

218 res = default 

219 else: 

220 raise 

221 except ValidationError as err: 

222 logger = logging.getLogger() 

223 logger.error(f"Failed parsing config key- {key}") 

224 raise 

225 except: 

226 raise 

227 return res 

228 

229 def __repr__(self) -> str: 

230 return json.dumps( 

231 {k: str(v.value) for k, v in self.entries.items()}, 

232 indent=2, 

233 sort_keys=True, 

234 ) 

235 

236 def debug_repr(self) -> str: 

237 """Repr() intended for debug purposes, since it runs repr() on each 

238 entry.value, it is more accurate than str(entry.value)""" 

239 repr_string = "{}(Confi):\n".format(self.__class__.__name__) 

240 items = list(self.entries.items()) 

241 items.sort(key=lambda item: item[0]) 

242 indent = " " * 4 

243 for key, entry in items: 

244 repr_string += f"{indent}{key}: {repr(entry.value)}\n" 

245 return repr_string 

246 

247 def get_cli_object( 

248 self, 

249 config_objects: List["Confi"] = None, 

250 typer_app: Typer = None, 

251 help: str = None, 

252 on_start: Callable = None, 

253 ): 

254 if config_objects is None: 

255 config_objects = [] 

256 config_objects.append(self) 

257 return get_cli_object_for_config_objects( 

258 config_objects, typer_app=typer_app, help=help, on_start=on_start 

259 ) 

260 

261 def cli( 

262 self, 

263 config_objects: List["Confi"] = None, 

264 typer_app: Typer = None, 

265 help: str = None, 

266 on_start: Callable = None, 

267 ): 

268 """Run a command-line-interface based on this configuration set, other 

269 config sets, and s typer cli app. 

270 

271 Args: 

272 config_objects (List[Confi, optional): additional config objects to share the CLI with this one. Defaults to None. 

273 typer_app (Typer, optional): A typer cli app with commands to expose to the CLI. Defaults to None. 

274 """ 

275 self.get_cli_object( 

276 config_objects, typer_app=typer_app, help=help, on_start=on_start 

277 )() 

278 

279 def on_load(self): 

280 """Callback called upon configuration load Add dynamic values you want 

281 set here (i.e. values which are based on other values)""" 

282 pass 

283 

284 def __setattr__(self, name: str, value: Any) -> None: 

285 """Make sure value updates are saved in internal entries as well.""" 

286 super().__setattr__(name, value) 

287 # update entry as well (to sync with CLI, etc. ) 

288 if not name.startswith("_") and name in self._entries: 

289 self._entries[name].value = value 

290 

291 def delay(self, value): 

292 delayed_entry = ConfiDelay(value, index=self._counter) 

293 self._counter += 1 

294 return delayed_entry 

295 

296 # -- parser setters -- 

297 

298 def str(self, key, default=undefined, description=None, **kwargs) -> str: 

299 return self._process( 

300 key, description=description, default=default, type=str, **kwargs 

301 ) 

302 

303 def int(self, key, default=undefined, description=None, **kwargs) -> int: 

304 return self._process( 

305 key, 

306 description=description, 

307 default=default, 

308 cast=int, 

309 type=int, 

310 **kwargs, 

311 ) 

312 

313 def bool(self, key, default=undefined, description=None, **kwargs) -> bool: 

314 return self._process( 

315 key, 

316 description=description, 

317 default=default, 

318 cast=cast_boolean, 

319 type=bool, 

320 **kwargs, 

321 ) 

322 

323 def float(self, key, default=undefined, description=None, **kwargs) -> float: 

324 return self._process( 

325 key, 

326 description=description, 

327 default=default, 

328 cast=float, 

329 type=float, 

330 **kwargs, 

331 ) 

332 

333 def list( 

334 self, 

335 key, 

336 default=undefined, 

337 sub_cast=text_type, 

338 delimiter=",", 

339 strip=string.whitespace, 

340 description=None, 

341 **kwargs, 

342 ) -> list: 

343 return self._process( 

344 key, 

345 default=default, 

346 description=description, 

347 cast=Csv(cast=sub_cast, delimiter=delimiter, strip=strip), 

348 type=list, 

349 **kwargs, 

350 ) 

351 

352 def model( 

353 self, key, model_type: T, default=undefined, description=None, **kwargs 

354 ) -> T: 

355 """Parse a config using a Pydantic model.""" 

356 x = self._process( 

357 key, 

358 description=description, 

359 default=default, 

360 cast=cast_pydantic(model_type), 

361 cast_from_json=cast_pydantic(model_type), 

362 type=model_type, 

363 **kwargs, 

364 ) 

365 return x 

366 

367 def enum( 

368 self, 

369 key, 

370 enum_type: EnumT, 

371 default=undefined, 

372 description=None, 

373 **kwargs, 

374 ) -> EnumT: 

375 return self._process( 

376 key, 

377 description=description, 

378 default=default, 

379 cast=enum_type, 

380 cast_from_json=enum_type, 

381 type=enum_type, 

382 **kwargs, 

383 ) 

384 

385 @log_exception() 

386 def private_key( 

387 self, 

388 key: str, 

389 default: Any = undefined, 

390 description: str = None, 

391 key_format: Optional[EncryptionKeyFormat] = None, 

392 passphrase: Optional[str] = None, 

393 **kwargs, 

394 ) -> Optional[PrivateKey]: 

395 """Parse a cryptographic private key from env vars.""" 

396 cast_key = partial( 

397 cast_private_key, key_format=key_format, passphrase=passphrase 

398 ) 

399 return self._process( 

400 key, 

401 description=description, 

402 default=default, 

403 cast=cast_key, 

404 cast_from_json=cast_key, 

405 type=PrivateKey, 

406 **kwargs, 

407 ) 

408 

409 @log_exception() 

410 def public_key( 

411 self, 

412 key: str, 

413 default: Any = undefined, 

414 description: str = None, 

415 key_format: Optional[EncryptionKeyFormat] = None, 

416 **kwargs, 

417 ) -> Optional[PublicKey]: 

418 """Parse a cryptographic public key from env vars.""" 

419 cast_key = partial(cast_public_key, key_format=key_format) 

420 return self._process( 

421 key, 

422 description=description, 

423 default=default, 

424 cast=cast_key, 

425 cast_from_json=cast_key, 

426 type=PublicKey, 

427 **kwargs, 

428 ) 

429 

430 

431# default parser 

432confi = Confi()