Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/azure/core/settings.py: 43%
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
31import logging
32import os
33from typing import (
34 Type,
35 Optional,
36 Callable,
37 Union,
38 Dict,
39 Any,
40 TypeVar,
41 Tuple,
42 Generic,
43 Mapping,
44 List,
45 TYPE_CHECKING,
46)
47from ._azure_clouds import AzureClouds
49if TYPE_CHECKING:
50 from azure.core.tracing import AbstractSpan
52ValidInputType = TypeVar("ValidInputType")
53ValueType = TypeVar("ValueType")
56__all__ = ("settings", "Settings")
59# https://www.python.org/dev/peps/pep-0484/#support-for-singleton-types-in-unions
60class _Unset(Enum):
61 token = 0
64_unset = _Unset.token
67def convert_bool(value: Union[str, bool]) -> bool:
68 """Convert a string to True or False
70 If a boolean is passed in, it is returned as-is. Otherwise the function
71 maps the following strings, ignoring case:
73 * "yes", "1", "on" -> True
74 " "no", "0", "off" -> False
76 :param value: the value to convert
77 :type value: str or bool
78 :returns: A boolean value matching the intent of the input
79 :rtype: bool
80 :raises ValueError: If conversion to bool fails
82 """
83 if isinstance(value, bool):
84 return value
85 val = value.lower()
86 if val in ["yes", "1", "on", "true", "True"]:
87 return True
88 if val in ["no", "0", "off", "false", "False"]:
89 return False
90 raise ValueError("Cannot convert {} to boolean value".format(value))
93def convert_tracing_enabled(value: Optional[Union[str, bool]]) -> bool:
94 """Convert tracing value to bool with regard to tracing implementation.
96 :param value: the value to convert
97 :type value: str or bool or None
98 :returns: A boolean value matching the intent of the input
99 :rtype: bool
100 :raises ValueError: If conversion to bool fails
101 """
102 if value is None:
103 # If tracing_enabled was not explicitly set to a boolean, determine tracing enablement
104 # based on tracing_implementation being set.
105 if settings.tracing_implementation():
106 return True
107 return False
108 return convert_bool(value)
111_levels = {
112 "CRITICAL": logging.CRITICAL,
113 "ERROR": logging.ERROR,
114 "WARNING": logging.WARNING,
115 "INFO": logging.INFO,
116 "DEBUG": logging.DEBUG,
117}
120def convert_logging(value: Union[str, int]) -> int:
121 """Convert a string to a Python logging level
123 If a log level is passed in, it is returned as-is. Otherwise the function
124 understands the following strings, ignoring case:
126 * "critical"
127 * "error"
128 * "warning"
129 * "info"
130 * "debug"
132 :param value: the value to convert
133 :type value: str or int
134 :returns: A log level as an int. See the logging module for details.
135 :rtype: int
136 :raises ValueError: If conversion to log level fails
138 """
139 if isinstance(value, int):
140 # If it's an int, return it. We don't need to check if it's in _levels, as custom int levels are allowed.
141 # https://docs.python.org/3/library/logging.html#levels
142 return value
143 val = value.upper()
144 level = _levels.get(val)
145 if not level:
146 raise ValueError("Cannot convert {} to log level, valid values are: {}".format(value, ", ".join(_levels)))
147 return level
150def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:
151 """Convert a string to an Azure Cloud
153 :param value: the value to convert
154 :type value: string
155 :returns: An AzureClouds enum value
156 :rtype: AzureClouds
157 :raises ValueError: If conversion to AzureClouds fails
159 """
160 if isinstance(value, AzureClouds):
161 return value
162 if isinstance(value, str):
163 azure_clouds = {cloud.name: cloud for cloud in AzureClouds}
164 if value in azure_clouds:
165 return azure_clouds[value]
166 raise ValueError(
167 "Cannot convert {} to Azure Cloud, valid values are: {}".format(value, ", ".join(azure_clouds.keys()))
168 )
169 raise ValueError("Cannot convert {} to Azure Cloud".format(value))
172def _get_opencensus_span() -> Optional[Type[AbstractSpan]]:
173 """Returns the OpenCensusSpan if the opencensus tracing plugin is installed else returns None.
175 :rtype: type[AbstractSpan] or None
176 :returns: OpenCensusSpan type or None
177 """
178 try:
179 from azure.core.tracing.ext.opencensus_span import (
180 OpenCensusSpan,
181 )
183 return OpenCensusSpan
184 except ImportError:
185 return None
188def _get_opentelemetry_span() -> Optional[Type[AbstractSpan]]:
189 """Returns the OpenTelemetrySpan if the opentelemetry tracing plugin is installed else returns None.
191 :rtype: type[AbstractSpan] or None
192 :returns: OpenTelemetrySpan type or None
193 """
194 try:
195 from azure.core.tracing.ext.opentelemetry_span import (
196 OpenTelemetrySpan,
197 )
199 return OpenTelemetrySpan
200 except ImportError:
201 return None
204_tracing_implementation_dict: Dict[str, Callable[[], Optional[Type[AbstractSpan]]]] = {
205 "opencensus": _get_opencensus_span,
206 "opentelemetry": _get_opentelemetry_span,
207}
210def convert_tracing_impl(value: Optional[Union[str, Type[AbstractSpan]]]) -> Optional[Type[AbstractSpan]]:
211 """Convert a string to AbstractSpan
213 If a AbstractSpan is passed in, it is returned as-is. Otherwise the function
214 understands the following strings, ignoring case:
216 * "opencensus"
217 * "opentelemetry"
219 :param value: the value to convert
220 :type value: string
221 :returns: AbstractSpan
222 :raises ValueError: If conversion to AbstractSpan fails
224 """
225 if value is None:
226 return None
228 if not isinstance(value, str):
229 return value
231 value = value.lower()
232 get_wrapper_class = _tracing_implementation_dict.get(value, lambda: _unset)
233 wrapper_class: Optional[Union[_Unset, Type[AbstractSpan]]] = get_wrapper_class()
234 if wrapper_class is _unset:
235 raise ValueError(
236 "Cannot convert {} to AbstractSpan, valid values are: {}".format(
237 value, ", ".join(_tracing_implementation_dict)
238 )
239 )
240 return wrapper_class
243class PrioritizedSetting(Generic[ValidInputType, ValueType]):
244 """Return a value for a global setting according to configuration precedence.
246 The following methods are searched in order for the setting:
248 4. immediate values
249 3. previously user-set value
250 2. environment variable
251 1. system setting
252 0. implicit default
254 If a value cannot be determined, a RuntimeError is raised.
256 The ``env_var`` argument specifies the name of an environment to check for
257 setting values, e.g. ``"AZURE_LOG_LEVEL"``.
258 If a ``convert`` function is provided, the result will be converted before being used.
260 The optional ``system_hook`` can be used to specify a function that will
261 attempt to look up a value for the setting from system-wide configurations.
262 If a ``convert`` function is provided, the hook result will be converted before being used.
264 The optional ``default`` argument specified an implicit default value for
265 the setting that is returned if no other methods provide a value. If a ``convert`` function is provided,
266 ``default`` will be converted before being used.
268 A ``convert`` argument may be provided to convert values before they are
269 returned. For instance to concert log levels in environment variables
270 to ``logging`` module values. If a ``convert`` function is provided, it must support
271 str as valid input type.
273 :param str name: the name of the setting
274 :param str env_var: the name of an environment variable to check for the setting
275 :param callable system_hook: a function that will attempt to look up a value for the setting
276 :param default: an implicit default value for the setting
277 :type default: any
278 :param callable convert: a function to convert values before they are returned
279 """
281 def __init__(
282 self,
283 name: str,
284 env_var: Optional[str] = None,
285 system_hook: Optional[Callable[[], ValidInputType]] = None,
286 default: Union[ValidInputType, _Unset] = _unset,
287 convert: Optional[Callable[[Union[ValidInputType, str]], ValueType]] = None,
288 ):
290 self._name = name
291 self._env_var = env_var
292 self._system_hook = system_hook
293 self._default = default
294 noop_convert: Callable[[Any], Any] = lambda x: x
295 self._convert: Callable[[Union[ValidInputType, str]], ValueType] = convert if convert else noop_convert
296 self._user_value: Union[ValidInputType, _Unset] = _unset
298 def __repr__(self) -> str:
299 return "PrioritizedSetting(%r)" % self._name
301 def __call__(self, value: Optional[ValidInputType] = None) -> ValueType:
302 """Return the setting value according to the standard precedence.
304 :param value: value
305 :type value: str or int or float or None
306 :returns: the value of the setting
307 :rtype: str or int or float
308 :raises RuntimeError: if no value can be determined
309 """
311 # 4. immediate values
312 if value is not None:
313 return self._convert(value)
315 # 3. previously user-set value
316 if not isinstance(self._user_value, _Unset):
317 return self._convert(self._user_value)
319 # 2. environment variable
320 if self._env_var and self._env_var in os.environ:
321 return self._convert(os.environ[self._env_var])
323 # 1. system setting
324 if self._system_hook:
325 return self._convert(self._system_hook())
327 # 0. implicit default
328 if not isinstance(self._default, _Unset):
329 return self._convert(self._default)
331 raise RuntimeError("No configured value found for setting %r" % self._name)
333 def __get__(self, instance: Any, owner: Optional[Any] = None) -> PrioritizedSetting[ValidInputType, ValueType]:
334 return self
336 def __set__(self, instance: Any, value: ValidInputType) -> None:
337 self.set_value(value)
339 def set_value(self, value: ValidInputType) -> None:
340 """Specify a value for this setting programmatically.
342 A value set this way takes precedence over all other methods except
343 immediate values.
345 :param value: a user-set value for this setting
346 :type value: str or int or float
347 """
348 self._user_value = value
350 def unset_value(self) -> None:
351 """Unset the previous user value such that the priority is reset."""
352 self._user_value = _unset
354 @property
355 def env_var(self) -> Optional[str]:
356 return self._env_var
358 @property
359 def default(self) -> Union[ValidInputType, _Unset]:
360 return self._default
363class Settings:
364 """Settings for globally used Azure configuration values.
366 You probably don't want to create an instance of this class, but call the singleton instance:
368 .. code-block:: python
370 from azure.core.settings import settings
371 settings.log_level = log_level = logging.DEBUG
373 The following methods are searched in order for a setting:
375 4. immediate values
376 3. previously user-set value
377 2. environment variable
378 1. system setting
379 0. implicit default
381 An implicit default is (optionally) defined by the setting attribute itself.
383 A system setting value can be obtained from registries or other OS configuration
384 for settings that support that method.
386 An environment variable value is obtained from ``os.environ``
388 User-set values many be specified by assigning to the attribute:
390 .. code-block:: python
392 settings.log_level = log_level = logging.DEBUG
394 Immediate values are (optionally) provided when the setting is retrieved:
396 .. code-block:: python
398 settings.log_level(logging.DEBUG())
400 Immediate values are most often useful to provide from optional arguments
401 to client functions. If the argument value is not None, it will be returned
402 as-is. Otherwise, the setting searches other methods according to the
403 precedence rules.
405 Immutable configuration snapshots can be created with the following methods:
407 * settings.defaults returns the base defaultsvalues , ignoring any environment or system
408 or user settings
410 * settings.current returns the current computation of settings including prioritization
411 of configuration sources, unless defaults_only is set to True (in which case the result
412 is identical to settings.defaults)
414 * settings.config can be called with specific values to override what settings.current
415 would provide
417 .. code-block:: python
419 # return current settings with log level overridden
420 settings.config(log_level=logging.DEBUG)
422 :cvar log_level: a log level to use across all Azure client SDKs (AZURE_LOG_LEVEL)
423 :type log_level: PrioritizedSetting
424 :cvar tracing_enabled: Whether tracing should be enabled across Azure SDKs (AZURE_TRACING_ENABLED)
425 :type tracing_enabled: PrioritizedSetting
426 :cvar tracing_implementation: The tracing implementation to use (AZURE_SDK_TRACING_IMPLEMENTATION)
427 :type tracing_implementation: PrioritizedSetting
429 :Example:
431 >>> import logging
432 >>> from azure.core.settings import settings
433 >>> settings.log_level = logging.DEBUG
434 >>> settings.log_level()
435 10
437 >>> settings.log_level(logging.WARN)
438 30
440 """
442 def __init__(self) -> None:
443 self._defaults_only: bool = False
445 @property
446 def defaults_only(self) -> bool:
447 """Whether to ignore environment and system settings and return only base default values.
449 :rtype: bool
450 :returns: Whether to ignore environment and system settings and return only base default values.
451 """
452 return self._defaults_only
454 @defaults_only.setter
455 def defaults_only(self, value: bool) -> None:
456 self._defaults_only = value
458 @property
459 def defaults(self) -> Tuple[Any, ...]:
460 """Return implicit default values for all settings, ignoring environment and system.
462 :rtype: namedtuple
463 :returns: The implicit default values for all settings
464 """
465 props = {k: v.default for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
466 return self._config(props)
468 @property
469 def current(self) -> Tuple[Any, ...]:
470 """Return the current values for all settings.
472 :rtype: namedtuple
473 :returns: The current values for all settings
474 """
475 if self.defaults_only:
476 return self.defaults
477 return self.config()
479 def config(self, **kwargs: Any) -> Tuple[Any, ...]:
480 """Return the currently computed settings, with values overridden by parameter values.
482 :rtype: namedtuple
483 :returns: The current values for all settings, with values overridden by parameter values
485 Examples:
487 .. code-block:: python
489 # return current settings with log level overridden
490 settings.config(log_level=logging.DEBUG)
492 """
493 props = {k: v() for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
494 props.update(kwargs)
495 return self._config(props)
497 def _config(self, props: Mapping[str, Any]) -> Tuple[Any, ...]:
498 keys: List[str] = list(props.keys())
499 # https://github.com/python/mypy/issues/4414
500 Config = namedtuple("Config", keys) # type: ignore
501 return Config(**props)
503 log_level: PrioritizedSetting[Union[str, int], int] = PrioritizedSetting(
504 "log_level",
505 env_var="AZURE_LOG_LEVEL",
506 convert=convert_logging,
507 default=logging.INFO,
508 )
510 tracing_enabled: PrioritizedSetting[Optional[Union[str, bool]], bool] = PrioritizedSetting(
511 "tracing_enabled",
512 env_var="AZURE_TRACING_ENABLED",
513 convert=convert_tracing_enabled,
514 default=None,
515 )
517 tracing_implementation: PrioritizedSetting[
518 Optional[Union[str, Type[AbstractSpan]]], Optional[Type[AbstractSpan]]
519 ] = PrioritizedSetting(
520 "tracing_implementation",
521 env_var="AZURE_SDK_TRACING_IMPLEMENTATION",
522 convert=convert_tracing_impl,
523 default=None,
524 )
526 azure_cloud: PrioritizedSetting[Union[str, AzureClouds], AzureClouds] = PrioritizedSetting(
527 "azure_cloud",
528 env_var="AZURE_CLOUD",
529 convert=convert_azure_cloud,
530 default=AzureClouds.AZURE_PUBLIC_CLOUD,
531 )
534settings: Settings = Settings()
535"""The settings unique instance.
537:type settings: Settings
538"""