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

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

5# The MIT License (MIT)

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

36from ._azure_clouds import AzureClouds

38ValidInputType = TypeVar("ValidInputType")

39ValueType = TypeVar("ValueType")

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

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

46class _Unset(Enum):

47 token = 0

50_unset = _Unset.token

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

54 """Convert a string to True or False

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

57 maps the following strings, ignoring case:

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

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

62 :param value: the value to convert

63 :type value: str or bool

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

65 :rtype: bool

66 :raises ValueError: If conversion to bool fails

68 """

69 if isinstance(value, bool):

70 return value

71 val = value.lower()

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

73 return True

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

75 return False

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

79_levels = {

80 "CRITICAL": logging.CRITICAL,

81 "ERROR": logging.ERROR,

82 "WARNING": logging.WARNING,

83 "INFO": logging.INFO,

84 "DEBUG": logging.DEBUG,

85}

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

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

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

92 understands the following strings, ignoring case:

94 * "critical"

95 * "error"

96 * "warning"

97 * "info"

98 * "debug"

100 :param value: the value to convert

101 :type value: str or int

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

103 :rtype: int

104 :raises ValueError: If conversion to log level fails

105

106 """

107 if isinstance(value, int):

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

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

110 return value

111 val = value.upper()

112 level = _levels.get(val)

113 if not level:

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

115 return level

116

117

118def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:

119 """Convert a string to an Azure Cloud

120

121 :param value: the value to convert

122 :type value: string

123 :returns: An AzureClouds enum value

124 :rtype: AzureClouds

125 :raises ValueError: If conversion to AzureClouds fails

126

127 """

128 if isinstance(value, AzureClouds):

129 return value

130 if isinstance(value, str):

131 azure_clouds = {cloud.name: cloud for cloud in AzureClouds}

132 if value in azure_clouds:

133 return azure_clouds[value]

134 raise ValueError(

135 "Cannot convert {} to Azure Cloud, valid values are: {}".format(value, ", ".join(azure_clouds.keys()))

136 )

137 raise ValueError("Cannot convert {} to Azure Cloud".format(value))

138

139

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

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

142

143 :rtype: type[AbstractSpan] or None

144 :returns: OpenCensusSpan type or None

145 """

146 try:

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

148 OpenCensusSpan,

149 )

150

151 return OpenCensusSpan

152 except ImportError:

153 return None

154

155

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

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

158

159 :rtype: type[AbstractSpan] or None

160 :returns: OpenTelemetrySpan type or None

161 """

162 try:

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

164 OpenTelemetrySpan,

165 )

166

167 return OpenTelemetrySpan

168 except ImportError:

169 return None

170

171

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

173 if "opencensus" not in sys.modules:

174 return None

175 return _get_opencensus_span()

176

177

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

179 if "opentelemetry" not in sys.modules:

180 return None

181 return _get_opentelemetry_span()

182

183

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

185 "opencensus": _get_opencensus_span,

186 "opentelemetry": _get_opentelemetry_span,

187}

188

189

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

191 """Convert a string to AbstractSpan

192

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

194 understands the following strings, ignoring case:

195

196 * "opencensus"

197 * "opentelemetry"

198

199 :param value: the value to convert

200 :type value: string

201 :returns: AbstractSpan

202 :raises ValueError: If conversion to AbstractSpan fails

203

204 """

205 if value is None:

206 return (

207 _get_opentelemetry_span_if_opentelemetry_is_imported() or _get_opencensus_span_if_opencensus_is_imported()

208 )

209

210 if not isinstance(value, str):

211 return value

212

213 value = value.lower()

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

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

216 if wrapper_class is _unset:

217 raise ValueError(

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

219 value, ", ".join(_tracing_implementation_dict)

220 )

221 )

222 return wrapper_class

223

224

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

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

227

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

229

230 4. immediate values

231 3. previously user-set value

232 2. environment variable

233 1. system setting

234 0. implicit default

235

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

237

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

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

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

241

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

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

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

245

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

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

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

249

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

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

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

253 str as valid input type.

254

255 :param str name: the name of the setting

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

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

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

259 :type default: any

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

261 """

262

263 def __init__(

264 self,

265 name: str,

266 env_var: Optional[str] = None,

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

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

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

270 ):

271

272 self._name = name

273 self._env_var = env_var

274 self._system_hook = system_hook

275 self._default = default

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

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

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

279

280 def __repr__(self) -> str:

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

282

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

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

285

286 :param value: value

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

288 :returns: the value of the setting

289 :rtype: str or int or float

290 :raises: RuntimeError if no value can be determined

291 """

292

293 # 4. immediate values

294 if value is not None:

295 return self._convert(value)

296

297 # 3. previously user-set value

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

299 return self._convert(self._user_value)

300

301 # 2. environment variable

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

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

304

305 # 1. system setting

306 if self._system_hook:

307 return self._convert(self._system_hook())

308

309 # 0. implicit default

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

311 return self._convert(self._default)

312

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

314

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

316 return self

317

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

319 self.set_value(value)

320

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

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

323

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

325 immediate values.

326

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

328 :type value: str or int or float

329 """

330 self._user_value = value

331

332 def unset_value(self) -> None:

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

334 self._user_value = _unset

335

336 @property

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

338 return self._env_var

339

340 @property

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

342 return self._default

343

344

345class Settings:

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

347

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

349

350 .. code-block:: python

351

352 from azure.core.settings import settings

353 settings.log_level = log_level = logging.DEBUG

354

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

356

357 4. immediate values

358 3. previously user-set value

359 2. environment variable

360 1. system setting

361 0. implicit default

362

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

364

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

366 for settings that support that method.

367

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

369

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

371

372 .. code-block:: python

373

374 settings.log_level = log_level = logging.DEBUG

375

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

377

378 .. code-block:: python

379

380 settings.log_level(logging.DEBUG())

381

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

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

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

385 precedence rules.

386

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

388

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

390 or user settings

391

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

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

394 is identical to settings.defaults)

395

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

397 would provide

398

399 .. code-block:: python

400

401 # return current settings with log level overridden

402 settings.config(log_level=logging.DEBUG)

403

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

405 :type log_level: PrioritizedSetting

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

407 :type tracing_enabled: PrioritizedSetting

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

409 :type tracing_implementation: PrioritizedSetting

410

411 :Example:

412

413 >>> import logging

414 >>> from azure.core.settings import settings

415 >>> settings.log_level = logging.DEBUG

416 >>> settings.log_level()

417 10

418

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

420 30

421

422 """

423

424 def __init__(self) -> None:

425 self._defaults_only: bool = False

426

427 @property

428 def defaults_only(self) -> bool:

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

430

431 :rtype: bool

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

433 """

434 return self._defaults_only

435

436 @defaults_only.setter

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

438 self._defaults_only = value

439

440 @property

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

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

443

444 :rtype: namedtuple

445 :returns: The implicit default values for all settings

446 """

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

448 return self._config(props)

449

450 @property

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

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

453

454 :rtype: namedtuple

455 :returns: The current values for all settings

456 """

457 if self.defaults_only:

458 return self.defaults

459 return self.config()

460

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

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

463

464 :rtype: namedtuple

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

466

467 Examples:

468

469 .. code-block:: python

470

471 # return current settings with log level overridden

472 settings.config(log_level=logging.DEBUG)

473

474 """

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

476 props.update(kwargs)

477 return self._config(props)

478

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

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

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

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

483 return Config(**props)

484

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

486 "log_level",

487 env_var="AZURE_LOG_LEVEL",

488 convert=convert_logging,

489 default=logging.INFO,

490 )

491

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

493 "tracing_enabled",

494 env_var="AZURE_TRACING_ENABLED",

495 convert=convert_bool,

496 default=False,

497 )

498

499 tracing_implementation: PrioritizedSetting[

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

501 ] = PrioritizedSetting(

502 "tracing_implementation",

503 env_var="AZURE_SDK_TRACING_IMPLEMENTATION",

504 convert=convert_tracing_impl,

505 default=None,

506 )

507

508 azure_cloud: PrioritizedSetting[Union[str, AzureClouds], AzureClouds] = PrioritizedSetting(

509 "azure_cloud",

510 env_var="AZURE_CLOUD",

511 convert=convert_azure_cloud,

512 default=AzureClouds.AZURE_PUBLIC_CLOUD,

513 )

514

515

516settings: Settings = Settings()

517"""The settings unique instance.

518

519:type settings: Settings

520"""