two version of R2R are here HEAD master

1 files changed, 532 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/azure/core/settings.py b/.venv/lib/python3.12/site-packages/azure/core/settings.py
new file mode 100644
index 00000000..8a5c07a9
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/core/settings.py
@@ -0,0 +1,532 @@
+# --------------------------------------------------------------------------
+#
+# 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
+"""