# --------------------------------------------------------------------------
#
# Copyright (c) Microsoft Corporation. All rights reserved.
#
# The MIT License (MIT)
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the ""Software""), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# --------------------------------------------------------------------------
"""Provide access to settings for globally used Azure configuration values.
"""
from __future__ import annotations
from collections import namedtuple
from enum import Enum
import logging
import os
import sys
from typing import (
Type,
Optional,
Callable,
Union,
Dict,
Any,
TypeVar,
Tuple,
Generic,
Mapping,
List,
)
from azure.core.tracing import AbstractSpan
from ._azure_clouds import AzureClouds
ValidInputType = TypeVar("ValidInputType")
ValueType = TypeVar("ValueType")
__all__ = ("settings", "Settings")
# https://www.python.org/dev/peps/pep-0484/#support-for-singleton-types-in-unions
class _Unset(Enum):
token = 0
_unset = _Unset.token
def convert_bool(value: Union[str, bool]) -> bool:
"""Convert a string to True or False
If a boolean is passed in, it is returned as-is. Otherwise the function
maps the following strings, ignoring case:
* "yes", "1", "on" -> True
" "no", "0", "off" -> False
:param value: the value to convert
:type value: str or bool
:returns: A boolean value matching the intent of the input
:rtype: bool
:raises ValueError: If conversion to bool fails
"""
if isinstance(value, bool):
return value
val = value.lower()
if val in ["yes", "1", "on", "true", "True"]:
return True
if val in ["no", "0", "off", "false", "False"]:
return False
raise ValueError("Cannot convert {} to boolean value".format(value))
_levels = {
"CRITICAL": logging.CRITICAL,
"ERROR": logging.ERROR,
"WARNING": logging.WARNING,
"INFO": logging.INFO,
"DEBUG": logging.DEBUG,
}
def convert_logging(value: Union[str, int]) -> int:
"""Convert a string to a Python logging level
If a log level is passed in, it is returned as-is. Otherwise the function
understands the following strings, ignoring case:
* "critical"
* "error"
* "warning"
* "info"
* "debug"
:param value: the value to convert
:type value: str or int
:returns: A log level as an int. See the logging module for details.
:rtype: int
:raises ValueError: If conversion to log level fails
"""
if isinstance(value, int):
# If it's an int, return it. We don't need to check if it's in _levels, as custom int levels are allowed.
# https://docs.python.org/3/library/logging.html#levels
return value
val = value.upper()
level = _levels.get(val)
if not level:
raise ValueError("Cannot convert {} to log level, valid values are: {}".format(value, ", ".join(_levels)))
return level
def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:
"""Convert a string to an Azure Cloud
:param value: the value to convert
:type value: string
:returns: An AzureClouds enum value
:rtype: AzureClouds
:raises ValueError: If conversion to AzureClouds fails
"""
if isinstance(value, AzureClouds):
return value
if isinstance(value, str):
azure_clouds = {cloud.name: cloud for cloud in AzureClouds}
if value in azure_clouds:
return azure_clouds[value]
raise ValueError(
"Cannot convert {} to Azure Cloud, valid values are: {}".format(value, ", ".join(azure_clouds.keys()))
)
raise ValueError("Cannot convert {} to Azure Cloud".format(value))
def _get_opencensus_span() -> Optional[Type[AbstractSpan]]:
"""Returns the OpenCensusSpan if the opencensus tracing plugin is installed else returns None.
:rtype: type[AbstractSpan] or None
:returns: OpenCensusSpan type or None
"""
try:
from azure.core.tracing.ext.opencensus_span import (
OpenCensusSpan,
)
return OpenCensusSpan
except ImportError:
return None
def _get_opentelemetry_span() -> Optional[Type[AbstractSpan]]:
"""Returns the OpenTelemetrySpan if the opentelemetry tracing plugin is installed else returns None.
:rtype: type[AbstractSpan] or None
:returns: OpenTelemetrySpan type or None
"""
try:
from azure.core.tracing.ext.opentelemetry_span import (
OpenTelemetrySpan,
)
return OpenTelemetrySpan
except ImportError:
return None
def _get_opencensus_span_if_opencensus_is_imported() -> Optional[Type[AbstractSpan]]:
if "opencensus" not in sys.modules:
return None
return _get_opencensus_span()
def _get_opentelemetry_span_if_opentelemetry_is_imported() -> Optional[Type[AbstractSpan]]:
if "opentelemetry" not in sys.modules:
return None
return _get_opentelemetry_span()
_tracing_implementation_dict: Dict[str, Callable[[], Optional[Type[AbstractSpan]]]] = {
"opencensus": _get_opencensus_span,
"opentelemetry": _get_opentelemetry_span,
}
def convert_tracing_impl(value: Optional[Union[str, Type[AbstractSpan]]]) -> Optional[Type[AbstractSpan]]:
"""Convert a string to AbstractSpan
If a AbstractSpan is passed in, it is returned as-is. Otherwise the function
understands the following strings, ignoring case:
* "opencensus"
* "opentelemetry"
:param value: the value to convert
:type value: string
:returns: AbstractSpan
:raises ValueError: If conversion to AbstractSpan fails
"""
if value is None:
return (
_get_opentelemetry_span_if_opentelemetry_is_imported() or _get_opencensus_span_if_opencensus_is_imported()
)
if not isinstance(value, str):
return value
value = value.lower()
get_wrapper_class = _tracing_implementation_dict.get(value, lambda: _unset)
wrapper_class: Optional[Union[_Unset, Type[AbstractSpan]]] = get_wrapper_class()
if wrapper_class is _unset:
raise ValueError(
"Cannot convert {} to AbstractSpan, valid values are: {}".format(
value, ", ".join(_tracing_implementation_dict)
)
)
return wrapper_class
class PrioritizedSetting(Generic[ValidInputType, ValueType]):
"""Return a value for a global setting according to configuration precedence.
The following methods are searched in order for the setting:
4. immediate values
3. previously user-set value
2. environment variable
1. system setting
0. implicit default
If a value cannot be determined, a RuntimeError is raised.
The ``env_var`` argument specifies the name of an environment to check for
setting values, e.g. ``"AZURE_LOG_LEVEL"``.
If a ``convert`` function is provided, the result will be converted before being used.
The optional ``system_hook`` can be used to specify a function that will
attempt to look up a value for the setting from system-wide configurations.
If a ``convert`` function is provided, the hook result will be converted before being used.
The optional ``default`` argument specified an implicit default value for
the setting that is returned if no other methods provide a value. If a ``convert`` function is provided,
``default`` will be converted before being used.
A ``convert`` argument may be provided to convert values before they are
returned. For instance to concert log levels in environment variables
to ``logging`` module values. If a ``convert`` function is provided, it must support
str as valid input type.
:param str name: the name of the setting
:param str env_var: the name of an environment variable to check for the setting
:param callable system_hook: a function that will attempt to look up a value for the setting
:param default: an implicit default value for the setting
:type default: any
:param callable convert: a function to convert values before they are returned
"""
def __init__(
self,
name: str,
env_var: Optional[str] = None,
system_hook: Optional[Callable[[], ValidInputType]] = None,
default: Union[ValidInputType, _Unset] = _unset,
convert: Optional[Callable[[Union[ValidInputType, str]], ValueType]] = None,
):
self._name = name
self._env_var = env_var
self._system_hook = system_hook
self._default = default
noop_convert: Callable[[Any], Any] = lambda x: x
self._convert: Callable[[Union[ValidInputType, str]], ValueType] = convert if convert else noop_convert
self._user_value: Union[ValidInputType, _Unset] = _unset
def __repr__(self) -> str:
return "PrioritizedSetting(%r)" % self._name
def __call__(self, value: Optional[ValidInputType] = None) -> ValueType:
"""Return the setting value according to the standard precedence.
:param value: value
:type value: str or int or float or None
:returns: the value of the setting
:rtype: str or int or float
:raises: RuntimeError if no value can be determined
"""
# 4. immediate values
if value is not None:
return self._convert(value)
# 3. previously user-set value
if not isinstance(self._user_value, _Unset):
return self._convert(self._user_value)
# 2. environment variable
if self._env_var and self._env_var in os.environ:
return self._convert(os.environ[self._env_var])
# 1. system setting
if self._system_hook:
return self._convert(self._system_hook())
# 0. implicit default
if not isinstance(self._default, _Unset):
return self._convert(self._default)
raise RuntimeError("No configured value found for setting %r" % self._name)
def __get__(self, instance: Any, owner: Optional[Any] = None) -> PrioritizedSetting[ValidInputType, ValueType]:
return self
def __set__(self, instance: Any, value: ValidInputType) -> None:
self.set_value(value)
def set_value(self, value: ValidInputType) -> None:
"""Specify a value for this setting programmatically.
A value set this way takes precedence over all other methods except
immediate values.
:param value: a user-set value for this setting
:type value: str or int or float
"""
self._user_value = value
def unset_value(self) -> None:
"""Unset the previous user value such that the priority is reset."""
self._user_value = _unset
@property
def env_var(self) -> Optional[str]:
return self._env_var
@property
def default(self) -> Union[ValidInputType, _Unset]:
return self._default
class Settings:
"""Settings for globally used Azure configuration values.
You probably don't want to create an instance of this class, but call the singleton instance:
.. code-block:: python
from azure.core.settings import settings
settings.log_level = log_level = logging.DEBUG
The following methods are searched in order for a setting:
4. immediate values
3. previously user-set value
2. environment variable
1. system setting
0. implicit default
An implicit default is (optionally) defined by the setting attribute itself.
A system setting value can be obtained from registries or other OS configuration
for settings that support that method.
An environment variable value is obtained from ``os.environ``
User-set values many be specified by assigning to the attribute:
.. code-block:: python
settings.log_level = log_level = logging.DEBUG
Immediate values are (optionally) provided when the setting is retrieved:
.. code-block:: python
settings.log_level(logging.DEBUG())
Immediate values are most often useful to provide from optional arguments
to client functions. If the argument value is not None, it will be returned
as-is. Otherwise, the setting searches other methods according to the
precedence rules.
Immutable configuration snapshots can be created with the following methods:
* settings.defaults returns the base defaultsvalues , ignoring any environment or system
or user settings
* settings.current returns the current computation of settings including prioritization
of configuration sources, unless defaults_only is set to True (in which case the result
is identical to settings.defaults)
* settings.config can be called with specific values to override what settings.current
would provide
.. code-block:: python
# return current settings with log level overridden
settings.config(log_level=logging.DEBUG)
:cvar log_level: a log level to use across all Azure client SDKs (AZURE_LOG_LEVEL)
:type log_level: PrioritizedSetting
:cvar tracing_enabled: Whether tracing should be enabled across Azure SDKs (AZURE_TRACING_ENABLED)
:type tracing_enabled: PrioritizedSetting
:cvar tracing_implementation: The tracing implementation to use (AZURE_SDK_TRACING_IMPLEMENTATION)
:type tracing_implementation: PrioritizedSetting
:Example:
>>> import logging
>>> from azure.core.settings import settings
>>> settings.log_level = logging.DEBUG
>>> settings.log_level()
10
>>> settings.log_level(logging.WARN)
30
"""
def __init__(self) -> None:
self._defaults_only: bool = False
@property
def defaults_only(self) -> bool:
"""Whether to ignore environment and system settings and return only base default values.
:rtype: bool
:returns: Whether to ignore environment and system settings and return only base default values.
"""
return self._defaults_only
@defaults_only.setter
def defaults_only(self, value: bool) -> None:
self._defaults_only = value
@property
def defaults(self) -> Tuple[Any, ...]:
"""Return implicit default values for all settings, ignoring environment and system.
:rtype: namedtuple
:returns: The implicit default values for all settings
"""
props = {k: v.default for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
return self._config(props)
@property
def current(self) -> Tuple[Any, ...]:
"""Return the current values for all settings.
:rtype: namedtuple
:returns: The current values for all settings
"""
if self.defaults_only:
return self.defaults
return self.config()
def config(self, **kwargs: Any) -> Tuple[Any, ...]:
"""Return the currently computed settings, with values overridden by parameter values.
:rtype: namedtuple
:returns: The current values for all settings, with values overridden by parameter values
Examples:
.. code-block:: python
# return current settings with log level overridden
settings.config(log_level=logging.DEBUG)
"""
props = {k: v() for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
props.update(kwargs)
return self._config(props)
def _config(self, props: Mapping[str, Any]) -> Tuple[Any, ...]:
keys: List[str] = list(props.keys())
# https://github.com/python/mypy/issues/4414
Config = namedtuple("Config", keys) # type: ignore
return Config(**props)
log_level: PrioritizedSetting[Union[str, int], int] = PrioritizedSetting(
"log_level",
env_var="AZURE_LOG_LEVEL",
convert=convert_logging,
default=logging.INFO,
)
tracing_enabled: PrioritizedSetting[Union[str, bool], bool] = PrioritizedSetting(
"tracing_enabled",
env_var="AZURE_TRACING_ENABLED",
convert=convert_bool,
default=False,
)
tracing_implementation: PrioritizedSetting[
Optional[Union[str, Type[AbstractSpan]]], Optional[Type[AbstractSpan]]
] = PrioritizedSetting(
"tracing_implementation",
env_var="AZURE_SDK_TRACING_IMPLEMENTATION",
convert=convert_tracing_impl,
default=None,
)
azure_cloud: PrioritizedSetting[Union[str, AzureClouds], AzureClouds] = PrioritizedSetting(
"azure_cloud",
env_var="AZURE_CLOUD",
convert=convert_azure_cloud,
default=AzureClouds.AZURE_PUBLIC_CLOUD,
)
settings: Settings = Settings()
"""The settings unique instance.
:type settings: Settings
"""
