# --------------------------------------------------------------------------
#
# 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 collections import namedtuple
from enum import Enum
import logging
import os
import sys
import six
from azure.core.tracing import AbstractSpan
try:
from typing import Type, Optional, Dict, Callable, cast, TYPE_CHECKING
except ImportError:
TYPE_CHECKING = False
if TYPE_CHECKING:
from typing import Any, Union
try:
# pylint:disable=unused-import
from azure.core.tracing.ext.opencensus_span import OpenCensusSpan # pylint:disable=redefined-outer-name
except ImportError:
pass
__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):
# type: (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: string
:returns: int
:raises ValueError: If conversion to bool fails
"""
if value in (True, False):
return value # type: ignore
val = value.lower() # type: ignore
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):
# type: (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: string
:returns: int
:raises ValueError: If conversion to log level fails
"""
if value in set(_levels.values()):
return value # type: ignore
val = value.upper() # type: ignore
level = _levels.get(val)
if not level:
raise ValueError("Cannot convert {} to log level, valid values are: {}".format(value, ", ".join(_levels)))
return level
def get_opencensus_span():
# type: () -> Optional[Type[AbstractSpan]]
"""Returns the OpenCensusSpan if opencensus is installed else returns None"""
try:
from azure.core.tracing.ext.opencensus_span import OpenCensusSpan # pylint:disable=redefined-outer-name
return OpenCensusSpan # type: ignore
except ImportError:
return None
def get_opencensus_span_if_opencensus_is_imported():
# type: () -> Optional[Type[AbstractSpan]]
if "opencensus" not in sys.modules:
return None
return get_opencensus_span()
_tracing_implementation_dict = {
"opencensus": get_opencensus_span
} # type: Dict[str, Callable[[], Optional[Type[AbstractSpan]]]]
def convert_tracing_impl(value):
# type: (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"
:param value: the value to convert
:type value: string
:returns: AbstractSpan
:raises ValueError: If conversion to AbstractSpan fails
"""
if value is None:
return get_opencensus_span_if_opencensus_is_imported()
if not isinstance(value, six.string_types):
return value
value = cast(str, value) # mypy clarity
value = value.lower()
get_wrapper_class = _tracing_implementation_dict.get(value, lambda: _unset)
wrapper_class = get_wrapper_class() # type: Union[None, _Unset, Type[AbstractSpan]]
if wrapper_class is _unset:
raise ValueError(
"Cannot convert {} to AbstractSpan, valid values are: {}".format(
value, ", ".join(_tracing_implementation_dict)
)
)
# type ignored until https://github.com/python/mypy/issues/7279
return wrapper_class # type: ignore
class PrioritizedSetting(object):
"""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"``.
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.
The optional ``default`` argument specified an implicit default value for
the setting that is returned if no other methods provide a value.
A ``convert`` agument may be provided to convert values before they are
returned. For instance to concert log levels in environment variables
to ``logging`` module values.
"""
def __init__(self, name, env_var=None, system_hook=None, default=_Unset, convert=None):
self._name = name
self._env_var = env_var
self._system_hook = system_hook
self._default = default
self._convert = convert if convert else lambda x: x
self._user_value = _Unset
def __repr__(self):
# type () -> str
return "PrioritizedSetting(%r)" % self._name
def __call__(self, value=None):
# type: (Any) -> Any
"""Return the setting value according to the standard precedence.
:param time: value
:type time: str or int or float or None (default: None)
:returns: the value of the setting
:rtype: str or int or float
:raises: RuntimeError
"""
# 4. immediate values
if value is not None:
return self._convert(value)
# 3. previously user-set value
if self._user_value is not _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 self._default is not _Unset:
return self._convert(self._default)
raise RuntimeError("No configured value found for setting %r" % self._name)
def __get__(self, instance, owner):
return self
def __set__(self, instance, value):
self.set_value(value)
def set_value(self, value):
# type: (Any) -> None
"""Specify a value for this setting programmatically.
A value set this way takes precedence over all other methods except
immediate values.
:param time: a user-set value for this setting
:type time: str or int or float
:returns: None
"""
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):
return self._env_var
@property
def default(self):
return self._default
[docs]class Settings(object):
"""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.common.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 prioritizatiom
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):
self._defaults_only = False
@property
def defaults_only(self):
"""Whether to ignore environment and system settings and return only base default values.
:rtype: bool
"""
return self._defaults_only
@defaults_only.setter
def defaults_only(self, value):
self._defaults_only = value
@property
def defaults(self):
"""Return implicit default values for all settings, ignoring environment and system.
:rtype: namedtuple
"""
props = {k: v.default for (k, v) in self.__class__.__dict__.items() if isinstance(v, PrioritizedSetting)}
return self._config(props)
@property
def current(self):
"""Return the current values for all settings.
:rtype: namedtuple
"""
if self.defaults_only:
return self.defaults
return self.config()
[docs] def config(self, **kwargs):
""" Return the currently computed 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): # pylint: disable=no-self-use
Config = namedtuple("Config", list(props.keys()))
return Config(**props)
log_level = PrioritizedSetting(
"log_level", env_var="AZURE_LOG_LEVEL", convert=convert_logging, default=logging.INFO
)
tracing_enabled = PrioritizedSetting(
"tracing_enbled", env_var="AZURE_TRACING_ENABLED", convert=convert_bool, default=False
)
tracing_implementation = PrioritizedSetting(
"tracing_implementation", env_var="AZURE_SDK_TRACING_IMPLEMENTATION", convert=convert_tracing_impl, default=None
)
settings = Settings()
"""The settings unique instance.
:type settings: Settings
"""