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

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

31from functools import cache

32import logging

33import os

34from typing import (

35 Type,

36 Optional,

37 Callable,

38 Union,

39 Dict,

40 Any,

41 TypeVar,

42 Tuple,

43 Generic,

44 Mapping,

45 List,

46 TYPE_CHECKING,

47)

48from ._azure_clouds import AzureClouds

50if TYPE_CHECKING:

51 from azure.core.tracing import AbstractSpan

53ValidInputType = TypeVar("ValidInputType")

54ValueType = TypeVar("ValueType")

56_LOGGER = logging.getLogger(__name__)

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

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

63class _Unset(Enum):

64 token = 0

67_unset = _Unset.token

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

71 """Convert a string to True or False

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

74 maps the following strings, ignoring case:

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

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

79 :param value: the value to convert

80 :type value: str or bool

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

82 :rtype: bool

83 :raises ValueError: If conversion to bool fails

85 """

86 if isinstance(value, bool):

87 return value

88 val = value.lower()

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

90 return True

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

92 return False

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

96def convert_tracing_enabled(value: Optional[Union[str, bool]]) -> bool:

97 """Convert tracing value to bool with regard to tracing implementation.

99 If the value cannot be converted to a bool, a warning is logged and the

100 default behavior is used (auto-detect based on whether a tracing

101 implementation is configured).

102

103 :param value: the value to convert

104 :type value: str or bool or None

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

106 :rtype: bool

107 """

108 if value is None:

109 # If tracing_enabled was not explicitly set to a boolean, determine tracing enablement

110 # based on tracing_implementation being set.

111 if settings.tracing_implementation():

112 return True

113 return False

114 try:

115 return convert_bool(value)

116 except ValueError:

117 _LOGGER.warning(

118 "Invalid value %r for AZURE_TRACING_ENABLED; falling back to default behavior. "

119 "Valid values are: 'true'/'false', 'yes'/'no', '1'/'0', 'on'/'off' (case-insensitive).",

120 value,

121 )

122 if settings.tracing_implementation():

123 return True

124 return False

125

126

127_levels = {

128 "CRITICAL": logging.CRITICAL,

129 "ERROR": logging.ERROR,

130 "WARNING": logging.WARNING,

131 "INFO": logging.INFO,

132 "DEBUG": logging.DEBUG,

133}

134

135

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

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

138

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

140 understands the following strings, ignoring case:

141

142 * "critical"

143 * "error"

144 * "warning"

145 * "info"

146 * "debug"

147 * "verbose" (treated as "debug")

148

149 If the value cannot be converted, a warning is logged and ``logging.INFO``

150 is returned.

151

152 :param value: the value to convert

153 :type value: str or int

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

155 :rtype: int

156

157 """

158 if isinstance(value, int):

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

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

161 return value

162 val = value.upper()

163 if val == "VERBOSE":

164 val = "DEBUG"

165 level = _levels.get(val)

166 if level is None:

167 _LOGGER.warning(

168 "Invalid log level %r for AZURE_LOG_LEVEL; falling back to INFO. Valid values are: %s.",

169 value,

170 ", ".join(_levels),

171 )

172 return logging.INFO

173 return level

174

175

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

177 """Convert a string to an Azure Cloud

178

179 :param value: the value to convert

180 :type value: string

181 :returns: An AzureClouds enum value

182 :rtype: AzureClouds

183 :raises ValueError: If conversion to AzureClouds fails

184

185 """

186 if isinstance(value, AzureClouds):

187 return value

188 if isinstance(value, str):

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

190 if value in azure_clouds:

191 return azure_clouds[value]

192 raise ValueError(

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

194 )

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

196

197

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

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

200

201 :rtype: type[AbstractSpan] or None

202 :returns: OpenCensusSpan type or None

203 """

204 try:

205 from azure.core.tracing.ext.opencensus_span import (

206 OpenCensusSpan,

207 )

208

209 return OpenCensusSpan

210 except ImportError:

211 return None

212

213

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

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

216

217 :rtype: type[AbstractSpan] or None

218 :returns: OpenTelemetrySpan type or None

219 """

220 try:

221 from azure.core.tracing.ext.opentelemetry_span import (

222 OpenTelemetrySpan,

223 )

224

225 return OpenTelemetrySpan

226 except ImportError:

227 return None

228

229

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

231 "opencensus": _get_opencensus_span,

232 "opentelemetry": _get_opentelemetry_span,

233}

234

235

236@cache

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

238 """Convert a string to AbstractSpan

239

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

241 understands the following strings, ignoring case:

242

243 * "opencensus"

244 * "opentelemetry"

245

246 If the value cannot be converted, a warning is logged and ``None`` is returned.

247

248 :param value: the value to convert

249 :type value: string

250 :returns: AbstractSpan

251

252 """

253 if value is None:

254 return None

255

256 if not isinstance(value, str):

257 return value

258

259 value = value.lower()

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

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

262 if wrapper_class is _unset:

263 _LOGGER.warning(

264 "Invalid tracing implementation %r for AZURE_SDK_TRACING_IMPLEMENTATION; falling back to default (None). "

265 "Valid values are: %s.",

266 value,

267 ", ".join(_tracing_implementation_dict),

268 )

269 return None

270 return wrapper_class

271

272

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

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

275

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

277

278 4. immediate values

279 3. previously user-set value

280 2. environment variable

281 1. system setting

282 0. implicit default

283

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

285

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

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

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

289

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

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

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

293

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

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

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

297

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

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

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

301 str as valid input type.

302

303 :param str name: the name of the setting

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

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

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

307 :type default: any

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

309 """

310

311 def __init__(

312 self,

313 name: str,

314 env_var: Optional[str] = None,

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

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

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

318 ):

319

320 self._name = name

321 self._env_var = env_var

322 self._system_hook = system_hook

323 self._default = default

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

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

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

327

328 def __repr__(self) -> str:

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

330

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

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

333

334 :param value: value

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

336 :returns: the value of the setting

337 :rtype: str or int or float

338 :raises RuntimeError: if no value can be determined

339 """

340

341 # 4. immediate values

342 if value is not None:

343 return self._convert(value)

344

345 # 3. previously user-set value

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

347 return self._convert(self._user_value)

348

349 # 2. environment variable

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

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

352

353 # 1. system setting

354 if self._system_hook:

355 return self._convert(self._system_hook())

356

357 # 0. implicit default

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

359 return self._convert(self._default)

360

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

362

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

364 return self

365

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

367 self.set_value(value)

368

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

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

371

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

373 immediate values.

374

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

376 :type value: str or int or float

377 """

378 self._user_value = value

379

380 def unset_value(self) -> None:

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

382 self._user_value = _unset

383

384 @property

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

386 return self._env_var

387

388 @property

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

390 return self._default

391

392

393class Settings:

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

395

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

397

398 .. code-block:: python

399

400 from azure.core.settings import settings

401 settings.log_level = log_level = logging.DEBUG

402

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

404

405 4. immediate values

406 3. previously user-set value

407 2. environment variable

408 1. system setting

409 0. implicit default

410

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

412

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

414 for settings that support that method.

415

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

417

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

419

420 .. code-block:: python

421

422 settings.log_level = log_level = logging.DEBUG

423

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

425

426 .. code-block:: python

427

428 settings.log_level(logging.DEBUG())

429

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

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

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

433 precedence rules.

434

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

436

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

438 or user settings

439

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

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

442 is identical to settings.defaults)

443

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

445 would provide

446

447 .. code-block:: python

448

449 # return current settings with log level overridden

450 settings.config(log_level=logging.DEBUG)

451

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

453 :type log_level: PrioritizedSetting

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

455 :type tracing_enabled: PrioritizedSetting

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

457 :type tracing_implementation: PrioritizedSetting

458 :cvar azure_cloud: The Azure cloud environment to use (AZURE_SDK_CLOUD_CONF)

459 :type azure_cloud: PrioritizedSetting

460

461 The following environment variables are used by the settings:

462

463 * ``AZURE_LOG_LEVEL`` - Logging level. Accepted values: ``CRITICAL``, ``ERROR``, ``WARNING``,

464 ``INFO``, ``DEBUG``, ``VERBOSE`` (case-insensitive). ``VERBOSE`` is treated as ``DEBUG``.

465 If the value is invalid, a warning is logged and ``INFO`` is used. Default: ``INFO``.

466 * ``AZURE_TRACING_ENABLED`` - Enable/disable tracing. Accepted values: ``true``/``false``,

467 ``yes``/``no``, ``1``/``0``, ``on``/``off`` (case-insensitive). If the value is invalid,

468 a warning is logged and the default is used. Default: auto-detected based on whether a

469 tracing implementation is configured.

470 * ``AZURE_SDK_TRACING_IMPLEMENTATION`` - Tracing implementation. Accepted values:

471 ``opentelemetry``. If the value is invalid, a warning is logged and the default is used.

472 Default: None.

473 * ``AZURE_SDK_CLOUD_CONF`` - Azure cloud environment. Accepted values:

474 ``AZURE_PUBLIC_CLOUD``, ``AZURE_CHINA_CLOUD``, ``AZURE_US_GOVERNMENT``.

475 Default: ``AZURE_PUBLIC_CLOUD``.

476

477 :Example:

478

479 >>> import logging

480 >>> from azure.core.settings import settings

481 >>> settings.log_level = logging.DEBUG

482 >>> settings.log_level()

483 10

484

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

486 30

487

488 """

489

490 def __init__(self) -> None:

491 self._defaults_only: bool = False

492

493 @property

494 def defaults_only(self) -> bool:

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

496

497 :rtype: bool

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

499 """

500 return self._defaults_only

501

502 @defaults_only.setter

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

504 self._defaults_only = value

505

506 @property

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

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

509

510 :rtype: namedtuple

511 :returns: The implicit default values for all settings

512 """

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

514 return self._config(props)

515

516 @property

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

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

519

520 :rtype: namedtuple

521 :returns: The current values for all settings

522 """

523 if self.defaults_only:

524 return self.defaults

525 return self.config()

526

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

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

529

530 :rtype: namedtuple

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

532

533 Examples:

534

535 .. code-block:: python

536

537 # return current settings with log level overridden

538 settings.config(log_level=logging.DEBUG)

539

540 """

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

542 props.update(kwargs)

543 return self._config(props)

544

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

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

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

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

549 return Config(**props)

550

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

552 "log_level",

553 env_var="AZURE_LOG_LEVEL",

554 convert=convert_logging,

555 default=logging.INFO,

556 )

557

558 tracing_enabled: PrioritizedSetting[Optional[Union[str, bool]], bool] = PrioritizedSetting(

559 "tracing_enabled",

560 env_var="AZURE_TRACING_ENABLED",

561 convert=convert_tracing_enabled,

562 default=None,

563 )

564

565 tracing_implementation: PrioritizedSetting[

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

567 ] = PrioritizedSetting(

568 "tracing_implementation",

569 env_var="AZURE_SDK_TRACING_IMPLEMENTATION",

570 convert=convert_tracing_impl,

571 default=None,

572 )

573

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

575 "azure_cloud",

576 env_var="AZURE_SDK_CLOUD_CONF",

577 convert=convert_azure_cloud,

578 default=AzureClouds.AZURE_PUBLIC_CLOUD,

579 )

580

581

582settings: Settings = Settings()

583"""The settings unique instance.

584

585:type settings: Settings

586"""