Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/azure/core/settings.py: 46%

1# -------------------------------------------------------------------------- 

2# 

3# Copyright (c) Microsoft Corporation. All rights reserved. 

4# 

5# The MIT License (MIT) 

6# 

7# Permission is hereby granted, free of charge, to any person obtaining a copy 

8# of this software and associated documentation files (the ""Software""), to deal 

9# in the Software without restriction, including without limitation the rights 

10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

11# copies of the Software, and to permit persons to whom the Software is 

12# furnished to do so, subject to the following conditions: 

13# 

14# The above copyright notice and this permission notice shall be included in 

15# all copies or substantial portions of the Software. 

16# 

17# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 

23# THE SOFTWARE. 

24# 

25# -------------------------------------------------------------------------- 

26"""Provide access to settings for globally used Azure configuration values. 

27""" 

28from __future__ import annotations 

29from collections import namedtuple 

30from enum import Enum 

31import logging 

32import os 

33import sys 

34from typing import Type, Optional, Callable, Union, Dict, Any, TypeVar, Tuple, Generic, Mapping, List 

35from azure.core.tracing import AbstractSpan 

36 

37ValidInputType = TypeVar("ValidInputType") 

38ValueType = TypeVar("ValueType") 

39 

40 

41__all__ = ("settings", "Settings") 

42 

43 

44# https://www.python.org/dev/peps/pep-0484/#support-for-singleton-types-in-unions 

45class _Unset(Enum): 

46 token = 0 

47 

48 

49_unset = _Unset.token 

50 

51 

52def convert_bool(value: Union[str, bool]) -> bool: 

53 """Convert a string to True or False 

54 

55 If a boolean is passed in, it is returned as-is. Otherwise the function 

56 maps the following strings, ignoring case: 

57 

58 * "yes", "1", "on" -> True 

59 " "no", "0", "off" -> False 

60 

61 :param value: the value to convert 

62 :type value: str or bool 

63 :returns: A boolean value matching the intent of the input 

64 :rtype: bool 

65 :raises ValueError: If conversion to bool fails 

66 

67 """ 

68 if isinstance(value, bool): 

69 return value 

70 val = value.lower() 

71 if val in ["yes", "1", "on", "true", "True"]: 

72 return True 

73 if val in ["no", "0", "off", "false", "False"]: 

74 return False 

75 raise ValueError("Cannot convert {} to boolean value".format(value)) 

76 

77 

78_levels = { 

79 "CRITICAL": logging.CRITICAL, 

80 "ERROR": logging.ERROR, 

81 "WARNING": logging.WARNING, 

82 "INFO": logging.INFO, 

83 "DEBUG": logging.DEBUG, 

84} 

85 

86 

87def convert_logging(value: Union[str, int]) -> int: 

88 """Convert a string to a Python logging level 

89 

90 If a log level is passed in, it is returned as-is. Otherwise the function 

91 understands the following strings, ignoring case: 

92 

93 * "critical" 

94 * "error" 

95 * "warning" 

96 * "info" 

97 * "debug" 

98 

99 :param value: the value to convert 

100 :type value: str or int 

101 :returns: A log level as an int. See the logging module for details. 

102 :rtype: int 

103 :raises ValueError: If conversion to log level fails 

104 

105 """ 

106 if isinstance(value, int): 

107 # If it's an int, return it. We don't need to check if it's in _levels, as custom int levels are allowed. 

108 # https://docs.python.org/3/library/logging.html#levels 

109 return value 

110 val = value.upper() 

111 level = _levels.get(val) 

112 if not level: 

113 raise ValueError("Cannot convert {} to log level, valid values are: {}".format(value, ", ".join(_levels))) 

114 return level 

115 

116 

117def _get_opencensus_span() -> Optional[Type[AbstractSpan]]: 

118 """Returns the OpenCensusSpan if the opencensus tracing plugin is installed else returns None. 

119 

120 :rtype: type[AbstractSpan] or None 

121 :returns: OpenCensusSpan type or None 

122 """ 

123 try: 

124 from azure.core.tracing.ext.opencensus_span import ( # pylint:disable=redefined-outer-name 

125 OpenCensusSpan, 

126 ) 

127 

128 return OpenCensusSpan 

129 except ImportError: 

130 return None 

131 

132 

133def _get_opentelemetry_span() -> Optional[Type[AbstractSpan]]: 

134 """Returns the OpenTelemetrySpan if the opentelemetry tracing plugin is installed else returns None. 

135 

136 :rtype: type[AbstractSpan] or None 

137 :returns: OpenTelemetrySpan type or None 

138 """ 

139 try: 

140 from azure.core.tracing.ext.opentelemetry_span import ( # pylint:disable=redefined-outer-name 

141 OpenTelemetrySpan, 

142 ) 

143 

144 return OpenTelemetrySpan 

145 except ImportError: 

146 return None 

147 

148 

149def _get_opencensus_span_if_opencensus_is_imported() -> Optional[Type[AbstractSpan]]: 

150 if "opencensus" not in sys.modules: 

151 return None 

152 return _get_opencensus_span() 

153 

154 

155def _get_opentelemetry_span_if_opentelemetry_is_imported() -> Optional[Type[AbstractSpan]]: 

156 if "opentelemetry" not in sys.modules: 

157 return None 

158 return _get_opentelemetry_span() 

159 

160 

161_tracing_implementation_dict: Dict[str, Callable[[], Optional[Type[AbstractSpan]]]] = { 

162 "opencensus": _get_opencensus_span, 

163 "opentelemetry": _get_opentelemetry_span, 

164} 

165 

166 

167def convert_tracing_impl(value: Optional[Union[str, Type[AbstractSpan]]]) -> Optional[Type[AbstractSpan]]: 

168 """Convert a string to AbstractSpan 

169 

170 If a AbstractSpan is passed in, it is returned as-is. Otherwise the function 

171 understands the following strings, ignoring case: 

172 

173 * "opencensus" 

174 * "opentelemetry" 

175 

176 :param value: the value to convert 

177 :type value: string 

178 :returns: AbstractSpan 

179 :raises ValueError: If conversion to AbstractSpan fails 

180 

181 """ 

182 if value is None: 

183 return ( 

184 _get_opencensus_span_if_opencensus_is_imported() or _get_opentelemetry_span_if_opentelemetry_is_imported() 

185 ) 

186 

187 if not isinstance(value, str): 

188 return value 

189 

190 value = value.lower() 

191 get_wrapper_class = _tracing_implementation_dict.get(value, lambda: _unset) 

192 wrapper_class: Optional[Union[_Unset, Type[AbstractSpan]]] = get_wrapper_class() 

193 if wrapper_class is _unset: 

194 raise ValueError( 

195 "Cannot convert {} to AbstractSpan, valid values are: {}".format( 

196 value, ", ".join(_tracing_implementation_dict) 

197 ) 

198 ) 

199 return wrapper_class 

200 

201 

202class PrioritizedSetting(Generic[ValidInputType, ValueType]): 

203 """Return a value for a global setting according to configuration precedence. 

204 

205 The following methods are searched in order for the setting: 

206 

207 4. immediate values 

208 3. previously user-set value 

209 2. environment variable 

210 1. system setting 

211 0. implicit default 

212 

213 If a value cannot be determined, a RuntimeError is raised. 

214 

215 The ``env_var`` argument specifies the name of an environment to check for 

216 setting values, e.g. ``"AZURE_LOG_LEVEL"``. 

217 If a ``convert`` function is provided, the result will be converted before being used. 

218 

219 The optional ``system_hook`` can be used to specify a function that will 

220 attempt to look up a value for the setting from system-wide configurations. 

221 If a ``convert`` function is provided, the hook result will be converted before being used. 

222 

223 The optional ``default`` argument specified an implicit default value for 

224 the setting that is returned if no other methods provide a value. If a ``convert`` function is provided, 

225 ``default`` will be converted before being used. 

226 

227 A ``convert`` argument may be provided to convert values before they are 

228 returned. For instance to concert log levels in environment variables 

229 to ``logging`` module values. If a ``convert`` function is provided, it must support 

230 str as valid input type. 

231 

232 :param str name: the name of the setting 

233 :param str env_var: the name of an environment variable to check for the setting 

234 :param callable system_hook: a function that will attempt to look up a value for the setting 

235 :param default: an implicit default value for the setting 

236 :type default: any 

237 :param callable convert: a function to convert values before they are returned 

238 """ 

239 

240 def __init__( 

241 self, 

242 name: str, 

243 env_var: Optional[str] = None, 

244 system_hook: Optional[Callable[[], ValidInputType]] = None, 

245 default: Union[ValidInputType, _Unset] = _unset, 

246 convert: Optional[Callable[[Union[ValidInputType, str]], ValueType]] = None, 

247 ): 

248 

249 self._name = name 

250 self._env_var = env_var 

251 self._system_hook = system_hook 

252 self._default = default 

253 noop_convert: Callable[[Any], Any] = lambda x: x 

254 self._convert: Callable[[Union[ValidInputType, str]], ValueType] = convert if convert else noop_convert 

255 self._user_value: Union[ValidInputType, _Unset] = _unset 

256 

257 def __repr__(self) -> str: 

258 return "PrioritizedSetting(%r)" % self._name 

259 

260 def __call__(self, value: Optional[ValidInputType] = None) -> ValueType: 

261 """Return the setting value according to the standard precedence. 

262 

263 :param value: value 

264 :type value: str or int or float or None 

265 :returns: the value of the setting 

266 :rtype: str or int or float 

267 :raises: RuntimeError if no value can be determined 

268 """ 

269 

270 # 4. immediate values 

271 if value is not None: 

272 return self._convert(value) 

273 

274 # 3. previously user-set value 

275 if not isinstance(self._user_value, _Unset): 

276 return self._convert(self._user_value) 

277 

278 # 2. environment variable 

279 if self._env_var and self._env_var in os.environ: 

280 return self._convert(os.environ[self._env_var]) 

281 

282 # 1. system setting 

283 if self._system_hook: 

284 return self._convert(self._system_hook()) 

285 

286 # 0. implicit default 

287 if not isinstance(self._default, _Unset): 

288 return self._convert(self._default) 

289 

290 raise RuntimeError("No configured value found for setting %r" % self._name) 

291 

292 def __get__(self, instance: Any, owner: Optional[Any] = None) -> PrioritizedSetting[ValidInputType, ValueType]: 

293 return self 

294 

295 def __set__(self, instance: Any, value: ValidInputType) -> None: 

296 self.set_value(value) 

297 

298 def set_value(self, value: ValidInputType) -> None: 

299 """Specify a value for this setting programmatically. 

300 

301 A value set this way takes precedence over all other methods except 

302 immediate values. 

303 

304 :param value: a user-set value for this setting 

305 :type value: str or int or float 

306 """ 

307 self._user_value = value 

308 

309 def unset_value(self) -> None: 

310 """Unset the previous user value such that the priority is reset.""" 

311 self._user_value = _unset 

312 

313 @property 

314 def env_var(self) -> Optional[str]: 

315 return self._env_var 

316 

317 @property 

318 def default(self) -> Union[ValidInputType, _Unset]: 

319 return self._default 

320 

321 

322class Settings: 

323 """Settings for globally used Azure configuration values. 

324 

325 You probably don't want to create an instance of this class, but call the singleton instance: 

326 

327 .. code-block:: python 

328 

329 from azure.core.settings import settings 

330 settings.log_level = log_level = logging.DEBUG 

331 

332 The following methods are searched in order for a setting: 

333 

334 4. immediate values 

335 3. previously user-set value 

336 2. environment variable 

337 1. system setting 

338 0. implicit default 

339 

340 An implicit default is (optionally) defined by the setting attribute itself. 

341 

342 A system setting value can be obtained from registries or other OS configuration 

343 for settings that support that method. 

344 

345 An environment variable value is obtained from ``os.environ`` 

346 

347 User-set values many be specified by assigning to the attribute: 

348 

349 .. code-block:: python 

350 

351 settings.log_level = log_level = logging.DEBUG 

352 

353 Immediate values are (optionally) provided when the setting is retrieved: 

354 

355 .. code-block:: python 

356 

357 settings.log_level(logging.DEBUG()) 

358 

359 Immediate values are most often useful to provide from optional arguments 

360 to client functions. If the argument value is not None, it will be returned 

361 as-is. Otherwise, the setting searches other methods according to the 

362 precedence rules. 

363 

364 Immutable configuration snapshots can be created with the following methods: 

365 

366 * settings.defaults returns the base defaultsvalues , ignoring any environment or system 

367 or user settings 

368 

369 * settings.current returns the current computation of settings including prioritization 

370 of configuration sources, unless defaults_only is set to True (in which case the result 

371 is identical to settings.defaults) 

372 

373 * settings.config can be called with specific values to override what settings.current 

374 would provide 

375 

376 .. code-block:: python 

377 

378 # return current settings with log level overridden 

379 settings.config(log_level=logging.DEBUG) 

380 

381 :cvar log_level: a log level to use across all Azure client SDKs (AZURE_LOG_LEVEL) 

382 :type log_level: PrioritizedSetting 

383 :cvar tracing_enabled: Whether tracing should be enabled across Azure SDKs (AZURE_TRACING_ENABLED) 

384 :type tracing_enabled: PrioritizedSetting 

385 :cvar tracing_implementation: The tracing implementation to use (AZURE_SDK_TRACING_IMPLEMENTATION) 

386 :type tracing_implementation: PrioritizedSetting 

387 

388 :Example: 

389 

390 >>> import logging 

391 >>> from azure.core.settings import settings 

392 >>> settings.log_level = logging.DEBUG 

393 >>> settings.log_level() 

394 10 

395 

396 >>> settings.log_level(logging.WARN) 

397 30 

398 

399 """ 

400 

401 def __init__(self) -> None: 

402 self._defaults_only: bool = False 

403 

404 @property 

405 def defaults_only(self) -> bool: 

406 """Whether to ignore environment and system settings and return only base default values. 

407 

408 :rtype: bool 

409 :returns: Whether to ignore environment and system settings and return only base default values. 

410 """ 

411 return self._defaults_only 

412 

413 @defaults_only.setter 

414 def defaults_only(self, value: bool) -> None: 

415 self._defaults_only = value 

416 

417 @property 

418 def defaults(self) -> Tuple[Any, ...]: 

419 """Return implicit default values for all settings, ignoring environment and system. 

420 

421 :rtype: namedtuple 

422 :returns: The implicit default values for all settings 

423 """ 

424 props = {k: v.default for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)} 

425 return self._config(props) 

426 

427 @property 

428 def current(self) -> Tuple[Any, ...]: 

429 """Return the current values for all settings. 

430 

431 :rtype: namedtuple 

432 :returns: The current values for all settings 

433 """ 

434 if self.defaults_only: 

435 return self.defaults 

436 return self.config() 

437 

438 def config(self, **kwargs: Any) -> Tuple[Any, ...]: 

439 """Return the currently computed settings, with values overridden by parameter values. 

440 

441 :rtype: namedtuple 

442 :returns: The current values for all settings, with values overridden by parameter values 

443 

444 Examples: 

445 

446 .. code-block:: python 

447 

448 # return current settings with log level overridden 

449 settings.config(log_level=logging.DEBUG) 

450 

451 """ 

452 props = {k: v() for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)} 

453 props.update(kwargs) 

454 return self._config(props) 

455 

456 def _config(self, props: Mapping[str, Any]) -> Tuple[Any, ...]: 

457 keys: List[str] = list(props.keys()) 

458 # https://github.com/python/mypy/issues/4414 

459 Config = namedtuple("Config", keys) # type: ignore 

460 return Config(**props) 

461 

462 log_level: PrioritizedSetting[Union[str, int], int] = PrioritizedSetting( 

463 "log_level", 

464 env_var="AZURE_LOG_LEVEL", 

465 convert=convert_logging, 

466 default=logging.INFO, 

467 ) 

468 

469 tracing_enabled: PrioritizedSetting[Union[str, bool], bool] = PrioritizedSetting( 

470 "tracing_enabled", 

471 env_var="AZURE_TRACING_ENABLED", 

472 convert=convert_bool, 

473 default=False, 

474 ) 

475 

476 tracing_implementation: PrioritizedSetting[ 

477 Optional[Union[str, Type[AbstractSpan]]], Optional[Type[AbstractSpan]] 

478 ] = PrioritizedSetting( 

479 "tracing_implementation", 

480 env_var="AZURE_SDK_TRACING_IMPLEMENTATION", 

481 convert=convert_tracing_impl, 

482 default=None, 

483 ) 

484 

485 

486settings: Settings = Settings() 

487"""The settings unique instance. 

488 

489:type settings: Settings 

490"""