Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/azure/core/settings.py: 44%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
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")
57__all__ = ("settings", "Settings")
60# https://www.python.org/dev/peps/pep-0484/#support-for-singleton-types-in-unions
61class _Unset(Enum):
62 token = 0
65_unset = _Unset.token
68def convert_bool(value: Union[str, bool]) -> bool:
69 """Convert a string to True or False
71 If a boolean is passed in, it is returned as-is. Otherwise the function
72 maps the following strings, ignoring case:
74 * "yes", "1", "on" -> True
75 " "no", "0", "off" -> False
77 :param value: the value to convert
78 :type value: str or bool
79 :returns: A boolean value matching the intent of the input
80 :rtype: bool
81 :raises ValueError: If conversion to bool fails
83 """
84 if isinstance(value, bool):
85 return value
86 val = value.lower()
87 if val in ["yes", "1", "on", "true", "True"]:
88 return True
89 if val in ["no", "0", "off", "false", "False"]:
90 return False
91 raise ValueError("Cannot convert {} to boolean value".format(value))
94def convert_tracing_enabled(value: Optional[Union[str, bool]]) -> bool:
95 """Convert tracing value to bool with regard to tracing implementation.
97 :param value: the value to convert
98 :type value: str or bool or None
99 :returns: A boolean value matching the intent of the input
100 :rtype: bool
101 :raises ValueError: If conversion to bool fails
102 """
103 if value is None:
104 # If tracing_enabled was not explicitly set to a boolean, determine tracing enablement
105 # based on tracing_implementation being set.
106 if settings.tracing_implementation():
107 return True
108 return False
109 return convert_bool(value)
112_levels = {
113 "CRITICAL": logging.CRITICAL,
114 "ERROR": logging.ERROR,
115 "WARNING": logging.WARNING,
116 "INFO": logging.INFO,
117 "DEBUG": logging.DEBUG,
118}
121def convert_logging(value: Union[str, int]) -> int:
122 """Convert a string to a Python logging level
124 If a log level is passed in, it is returned as-is. Otherwise the function
125 understands the following strings, ignoring case:
127 * "critical"
128 * "error"
129 * "warning"
130 * "info"
131 * "debug"
133 :param value: the value to convert
134 :type value: str or int
135 :returns: A log level as an int. See the logging module for details.
136 :rtype: int
137 :raises ValueError: If conversion to log level fails
139 """
140 if isinstance(value, int):
141 # If it's an int, return it. We don't need to check if it's in _levels, as custom int levels are allowed.
142 # https://docs.python.org/3/library/logging.html#levels
143 return value
144 val = value.upper()
145 level = _levels.get(val)
146 if not level:
147 raise ValueError("Cannot convert {} to log level, valid values are: {}".format(value, ", ".join(_levels)))
148 return level
151def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:
152 """Convert a string to an Azure Cloud
154 :param value: the value to convert
155 :type value: string
156 :returns: An AzureClouds enum value
157 :rtype: AzureClouds
158 :raises ValueError: If conversion to AzureClouds fails
160 """
161 if isinstance(value, AzureClouds):
162 return value
163 if isinstance(value, str):
164 azure_clouds = {cloud.name: cloud for cloud in AzureClouds}
165 if value in azure_clouds:
166 return azure_clouds[value]
167 raise ValueError(
168 "Cannot convert {} to Azure Cloud, valid values are: {}".format(value, ", ".join(azure_clouds.keys()))
169 )
170 raise ValueError("Cannot convert {} to Azure Cloud".format(value))
173def _get_opencensus_span() -> Optional[Type[AbstractSpan]]:
174 """Returns the OpenCensusSpan if the opencensus tracing plugin is installed else returns None.
176 :rtype: type[AbstractSpan] or None
177 :returns: OpenCensusSpan type or None
178 """
179 try:
180 from azure.core.tracing.ext.opencensus_span import (
181 OpenCensusSpan,
182 )
184 return OpenCensusSpan
185 except ImportError:
186 return None
189def _get_opentelemetry_span() -> Optional[Type[AbstractSpan]]:
190 """Returns the OpenTelemetrySpan if the opentelemetry tracing plugin is installed else returns None.
192 :rtype: type[AbstractSpan] or None
193 :returns: OpenTelemetrySpan type or None
194 """
195 try:
196 from azure.core.tracing.ext.opentelemetry_span import (
197 OpenTelemetrySpan,
198 )
200 return OpenTelemetrySpan
201 except ImportError:
202 return None
205_tracing_implementation_dict: Dict[str, Callable[[], Optional[Type[AbstractSpan]]]] = {
206 "opencensus": _get_opencensus_span,
207 "opentelemetry": _get_opentelemetry_span,
208}
211@cache
212def convert_tracing_impl(value: Optional[Union[str, Type[AbstractSpan]]]) -> Optional[Type[AbstractSpan]]:
213 """Convert a string to AbstractSpan
215 If a AbstractSpan is passed in, it is returned as-is. Otherwise the function
216 understands the following strings, ignoring case:
218 * "opencensus"
219 * "opentelemetry"
221 :param value: the value to convert
222 :type value: string
223 :returns: AbstractSpan
224 :raises ValueError: If conversion to AbstractSpan fails
226 """
227 if value is None:
228 return None
230 if not isinstance(value, str):
231 return value
233 value = value.lower()
234 get_wrapper_class = _tracing_implementation_dict.get(value, lambda: _unset)
235 wrapper_class: Optional[Union[_Unset, Type[AbstractSpan]]] = get_wrapper_class()
236 if wrapper_class is _unset:
237 raise ValueError(
238 "Cannot convert {} to AbstractSpan, valid values are: {}".format(
239 value, ", ".join(_tracing_implementation_dict)
240 )
241 )
242 return wrapper_class
245class PrioritizedSetting(Generic[ValidInputType, ValueType]):
246 """Return a value for a global setting according to configuration precedence.
248 The following methods are searched in order for the setting:
250 4. immediate values
251 3. previously user-set value
252 2. environment variable
253 1. system setting
254 0. implicit default
256 If a value cannot be determined, a RuntimeError is raised.
258 The ``env_var`` argument specifies the name of an environment to check for
259 setting values, e.g. ``"AZURE_LOG_LEVEL"``.
260 If a ``convert`` function is provided, the result will be converted before being used.
262 The optional ``system_hook`` can be used to specify a function that will
263 attempt to look up a value for the setting from system-wide configurations.
264 If a ``convert`` function is provided, the hook result will be converted before being used.
266 The optional ``default`` argument specified an implicit default value for
267 the setting that is returned if no other methods provide a value. If a ``convert`` function is provided,
268 ``default`` will be converted before being used.
270 A ``convert`` argument may be provided to convert values before they are
271 returned. For instance to concert log levels in environment variables
272 to ``logging`` module values. If a ``convert`` function is provided, it must support
273 str as valid input type.
275 :param str name: the name of the setting
276 :param str env_var: the name of an environment variable to check for the setting
277 :param callable system_hook: a function that will attempt to look up a value for the setting
278 :param default: an implicit default value for the setting
279 :type default: any
280 :param callable convert: a function to convert values before they are returned
281 """
283 def __init__(
284 self,
285 name: str,
286 env_var: Optional[str] = None,
287 system_hook: Optional[Callable[[], ValidInputType]] = None,
288 default: Union[ValidInputType, _Unset] = _unset,
289 convert: Optional[Callable[[Union[ValidInputType, str]], ValueType]] = None,
290 ):
292 self._name = name
293 self._env_var = env_var
294 self._system_hook = system_hook
295 self._default = default
296 noop_convert: Callable[[Any], Any] = lambda x: x
297 self._convert: Callable[[Union[ValidInputType, str]], ValueType] = convert if convert else noop_convert
298 self._user_value: Union[ValidInputType, _Unset] = _unset
300 def __repr__(self) -> str:
301 return "PrioritizedSetting(%r)" % self._name
303 def __call__(self, value: Optional[ValidInputType] = None) -> ValueType:
304 """Return the setting value according to the standard precedence.
306 :param value: value
307 :type value: str or int or float or None
308 :returns: the value of the setting
309 :rtype: str or int or float
310 :raises RuntimeError: if no value can be determined
311 """
313 # 4. immediate values
314 if value is not None:
315 return self._convert(value)
317 # 3. previously user-set value
318 if not isinstance(self._user_value, _Unset):
319 return self._convert(self._user_value)
321 # 2. environment variable
322 if self._env_var and self._env_var in os.environ:
323 return self._convert(os.environ[self._env_var])
325 # 1. system setting
326 if self._system_hook:
327 return self._convert(self._system_hook())
329 # 0. implicit default
330 if not isinstance(self._default, _Unset):
331 return self._convert(self._default)
333 raise RuntimeError("No configured value found for setting %r" % self._name)
335 def __get__(self, instance: Any, owner: Optional[Any] = None) -> PrioritizedSetting[ValidInputType, ValueType]:
336 return self
338 def __set__(self, instance: Any, value: ValidInputType) -> None:
339 self.set_value(value)
341 def set_value(self, value: ValidInputType) -> None:
342 """Specify a value for this setting programmatically.
344 A value set this way takes precedence over all other methods except
345 immediate values.
347 :param value: a user-set value for this setting
348 :type value: str or int or float
349 """
350 self._user_value = value
352 def unset_value(self) -> None:
353 """Unset the previous user value such that the priority is reset."""
354 self._user_value = _unset
356 @property
357 def env_var(self) -> Optional[str]:
358 return self._env_var
360 @property
361 def default(self) -> Union[ValidInputType, _Unset]:
362 return self._default
365class Settings:
366 """Settings for globally used Azure configuration values.
368 You probably don't want to create an instance of this class, but call the singleton instance:
370 .. code-block:: python
372 from azure.core.settings import settings
373 settings.log_level = log_level = logging.DEBUG
375 The following methods are searched in order for a setting:
377 4. immediate values
378 3. previously user-set value
379 2. environment variable
380 1. system setting
381 0. implicit default
383 An implicit default is (optionally) defined by the setting attribute itself.
385 A system setting value can be obtained from registries or other OS configuration
386 for settings that support that method.
388 An environment variable value is obtained from ``os.environ``
390 User-set values many be specified by assigning to the attribute:
392 .. code-block:: python
394 settings.log_level = log_level = logging.DEBUG
396 Immediate values are (optionally) provided when the setting is retrieved:
398 .. code-block:: python
400 settings.log_level(logging.DEBUG())
402 Immediate values are most often useful to provide from optional arguments
403 to client functions. If the argument value is not None, it will be returned
404 as-is. Otherwise, the setting searches other methods according to the
405 precedence rules.
407 Immutable configuration snapshots can be created with the following methods:
409 * settings.defaults returns the base defaultsvalues , ignoring any environment or system
410 or user settings
412 * settings.current returns the current computation of settings including prioritization
413 of configuration sources, unless defaults_only is set to True (in which case the result
414 is identical to settings.defaults)
416 * settings.config can be called with specific values to override what settings.current
417 would provide
419 .. code-block:: python
421 # return current settings with log level overridden
422 settings.config(log_level=logging.DEBUG)
424 :cvar log_level: a log level to use across all Azure client SDKs (AZURE_LOG_LEVEL)
425 :type log_level: PrioritizedSetting
426 :cvar tracing_enabled: Whether tracing should be enabled across Azure SDKs (AZURE_TRACING_ENABLED)
427 :type tracing_enabled: PrioritizedSetting
428 :cvar tracing_implementation: The tracing implementation to use (AZURE_SDK_TRACING_IMPLEMENTATION)
429 :type tracing_implementation: PrioritizedSetting
431 :Example:
433 >>> import logging
434 >>> from azure.core.settings import settings
435 >>> settings.log_level = logging.DEBUG
436 >>> settings.log_level()
437 10
439 >>> settings.log_level(logging.WARN)
440 30
442 """
444 def __init__(self) -> None:
445 self._defaults_only: bool = False
447 @property
448 def defaults_only(self) -> bool:
449 """Whether to ignore environment and system settings and return only base default values.
451 :rtype: bool
452 :returns: Whether to ignore environment and system settings and return only base default values.
453 """
454 return self._defaults_only
456 @defaults_only.setter
457 def defaults_only(self, value: bool) -> None:
458 self._defaults_only = value
460 @property
461 def defaults(self) -> Tuple[Any, ...]:
462 """Return implicit default values for all settings, ignoring environment and system.
464 :rtype: namedtuple
465 :returns: The implicit default values for all settings
466 """
467 props = {k: v.default for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
468 return self._config(props)
470 @property
471 def current(self) -> Tuple[Any, ...]:
472 """Return the current values for all settings.
474 :rtype: namedtuple
475 :returns: The current values for all settings
476 """
477 if self.defaults_only:
478 return self.defaults
479 return self.config()
481 def config(self, **kwargs: Any) -> Tuple[Any, ...]:
482 """Return the currently computed settings, with values overridden by parameter values.
484 :rtype: namedtuple
485 :returns: The current values for all settings, with values overridden by parameter values
487 Examples:
489 .. code-block:: python
491 # return current settings with log level overridden
492 settings.config(log_level=logging.DEBUG)
494 """
495 props = {k: v() for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
496 props.update(kwargs)
497 return self._config(props)
499 def _config(self, props: Mapping[str, Any]) -> Tuple[Any, ...]:
500 keys: List[str] = list(props.keys())
501 # https://github.com/python/mypy/issues/4414
502 Config = namedtuple("Config", keys) # type: ignore
503 return Config(**props)
505 log_level: PrioritizedSetting[Union[str, int], int] = PrioritizedSetting(
506 "log_level",
507 env_var="AZURE_LOG_LEVEL",
508 convert=convert_logging,
509 default=logging.INFO,
510 )
512 tracing_enabled: PrioritizedSetting[Optional[Union[str, bool]], bool] = PrioritizedSetting(
513 "tracing_enabled",
514 env_var="AZURE_TRACING_ENABLED",
515 convert=convert_tracing_enabled,
516 default=None,
517 )
519 tracing_implementation: PrioritizedSetting[
520 Optional[Union[str, Type[AbstractSpan]]], Optional[Type[AbstractSpan]]
521 ] = PrioritizedSetting(
522 "tracing_implementation",
523 env_var="AZURE_SDK_TRACING_IMPLEMENTATION",
524 convert=convert_tracing_impl,
525 default=None,
526 )
528 azure_cloud: PrioritizedSetting[Union[str, AzureClouds], AzureClouds] = PrioritizedSetting(
529 "azure_cloud",
530 env_var="AZURE_CLOUD",
531 convert=convert_azure_cloud,
532 default=AzureClouds.AZURE_PUBLIC_CLOUD,
533 )
536settings: Settings = Settings()
537"""The settings unique instance.
539:type settings: Settings
540"""