two version of R2R are hereHEADmaster

1 files changed, 1061 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/botocore/configprovider.py b/.venv/lib/python3.12/site-packages/botocore/configprovider.py
new file mode 100644
index 00000000..55c4654a
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/botocore/configprovider.py
@@ -0,0 +1,1061 @@
+# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""This module contains the interface for controlling how configuration
+is loaded.
+"""
+
+import copy
+import logging
+import os
+
+from botocore import utils
+from botocore.exceptions import InvalidConfigError
+
+logger = logging.getLogger(__name__)
+
+
+#: A default dictionary that maps the logical names for session variables
+#: to the specific environment variables and configuration file names
+#: that contain the values for these variables.
+#: When creating a new Session object, you can pass in your own dictionary
+#: to remap the logical names or to add new logical names.  You can then
+#: get the current value for these variables by using the
+#: ``get_config_variable`` method of the :class:`botocore.session.Session`
+#: class.
+#: These form the keys of the dictionary.  The values in the dictionary
+#: are tuples of (<config_name>, <environment variable>, <default value>,
+#: <conversion func>).
+#: The conversion func is a function that takes the configuration value
+#: as an argument and returns the converted value.  If this value is
+#: None, then the configuration value is returned unmodified.  This
+#: conversion function can be used to type convert config values to
+#: values other than the default values of strings.
+#: The ``profile`` and ``config_file`` variables should always have a
+#: None value for the first entry in the tuple because it doesn't make
+#: sense to look inside the config file for the location of the config
+#: file or for the default profile to use.
+#: The ``config_name`` is the name to look for in the configuration file,
+#: the ``env var`` is the OS environment variable (``os.environ``) to
+#: use, and ``default_value`` is the value to use if no value is otherwise
+#: found.
+#: NOTE: Fixing the spelling of this variable would be a breaking change.
+#: Please leave as is.
+BOTOCORE_DEFAUT_SESSION_VARIABLES = {
+    # logical:  config_file, env_var, default_value, conversion_func
+    'profile': (None, ['AWS_DEFAULT_PROFILE', 'AWS_PROFILE'], None, None),
+    'region': ('region', 'AWS_DEFAULT_REGION', None, None),
+    'data_path': ('data_path', 'AWS_DATA_PATH', None, None),
+    'config_file': (None, 'AWS_CONFIG_FILE', '~/.aws/config', None),
+    'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE', None, None),
+    'api_versions': ('api_versions', None, {}, None),
+    # This is the shared credentials file amongst sdks.
+    'credentials_file': (
+        None,
+        'AWS_SHARED_CREDENTIALS_FILE',
+        '~/.aws/credentials',
+        None,
+    ),
+    # These variables only exist in the config file.
+    # This is the number of seconds until we time out a request to
+    # the instance metadata service.
+    'metadata_service_timeout': (
+        'metadata_service_timeout',
+        'AWS_METADATA_SERVICE_TIMEOUT',
+        1,
+        int,
+    ),
+    # This is the number of request attempts we make until we give
+    # up trying to retrieve data from the instance metadata service.
+    'metadata_service_num_attempts': (
+        'metadata_service_num_attempts',
+        'AWS_METADATA_SERVICE_NUM_ATTEMPTS',
+        1,
+        int,
+    ),
+    'ec2_metadata_service_endpoint': (
+        'ec2_metadata_service_endpoint',
+        'AWS_EC2_METADATA_SERVICE_ENDPOINT',
+        None,
+        None,
+    ),
+    'ec2_metadata_service_endpoint_mode': (
+        'ec2_metadata_service_endpoint_mode',
+        'AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE',
+        None,
+        None,
+    ),
+    'ec2_metadata_v1_disabled': (
+        'ec2_metadata_v1_disabled',
+        'AWS_EC2_METADATA_V1_DISABLED',
+        False,
+        utils.ensure_boolean,
+    ),
+    'imds_use_ipv6': (
+        'imds_use_ipv6',
+        'AWS_IMDS_USE_IPV6',
+        False,
+        utils.ensure_boolean,
+    ),
+    'use_dualstack_endpoint': (
+        'use_dualstack_endpoint',
+        'AWS_USE_DUALSTACK_ENDPOINT',
+        None,
+        utils.ensure_boolean,
+    ),
+    'use_fips_endpoint': (
+        'use_fips_endpoint',
+        'AWS_USE_FIPS_ENDPOINT',
+        None,
+        utils.ensure_boolean,
+    ),
+    'ignore_configured_endpoint_urls': (
+        'ignore_configured_endpoint_urls',
+        'AWS_IGNORE_CONFIGURED_ENDPOINT_URLS',
+        None,
+        utils.ensure_boolean,
+    ),
+    'parameter_validation': ('parameter_validation', None, True, None),
+    # Client side monitoring configurations.
+    # Note: These configurations are considered internal to botocore.
+    # Do not use them until publicly documented.
+    'csm_enabled': (
+        'csm_enabled',
+        'AWS_CSM_ENABLED',
+        False,
+        utils.ensure_boolean,
+    ),
+    'csm_host': ('csm_host', 'AWS_CSM_HOST', '127.0.0.1', None),
+    'csm_port': ('csm_port', 'AWS_CSM_PORT', 31000, int),
+    'csm_client_id': ('csm_client_id', 'AWS_CSM_CLIENT_ID', '', None),
+    # Endpoint discovery configuration
+    'endpoint_discovery_enabled': (
+        'endpoint_discovery_enabled',
+        'AWS_ENDPOINT_DISCOVERY_ENABLED',
+        'auto',
+        None,
+    ),
+    'retry_mode': ('retry_mode', 'AWS_RETRY_MODE', 'legacy', None),
+    'defaults_mode': ('defaults_mode', 'AWS_DEFAULTS_MODE', 'legacy', None),
+    # We can't have a default here for v1 because we need to defer to
+    # whatever the defaults are in _retry.json.
+    'max_attempts': ('max_attempts', 'AWS_MAX_ATTEMPTS', None, int),
+    'user_agent_appid': ('sdk_ua_app_id', 'AWS_SDK_UA_APP_ID', None, None),
+    'request_min_compression_size_bytes': (
+        'request_min_compression_size_bytes',
+        'AWS_REQUEST_MIN_COMPRESSION_SIZE_BYTES',
+        10240,
+        None,
+    ),
+    'disable_request_compression': (
+        'disable_request_compression',
+        'AWS_DISABLE_REQUEST_COMPRESSION',
+        False,
+        utils.ensure_boolean,
+    ),
+    'sigv4a_signing_region_set': (
+        'sigv4a_signing_region_set',
+        'AWS_SIGV4A_SIGNING_REGION_SET',
+        None,
+        None,
+    ),
+    'request_checksum_calculation': (
+        'request_checksum_calculation',
+        'AWS_REQUEST_CHECKSUM_CALCULATION',
+        "when_supported",
+        None,
+    ),
+    'response_checksum_validation': (
+        'response_checksum_validation',
+        'AWS_RESPONSE_CHECKSUM_VALIDATION',
+        "when_supported",
+        None,
+    ),
+    'account_id_endpoint_mode': (
+        'account_id_endpoint_mode',
+        'AWS_ACCOUNT_ID_ENDPOINT_MODE',
+        'preferred',
+        None,
+    ),
+    'disable_host_prefix_injection': (
+        'disable_host_prefix_injection',
+        'AWS_DISABLE_HOST_PREFIX_INJECTION',
+        None,
+        utils.ensure_boolean,
+    ),
+}
+
+# Evaluate AWS_STS_REGIONAL_ENDPOINTS settings
+try:
+    # This is not a public interface and is subject to abrupt breaking changes.
+    # Any usage is not advised or supported in external code bases.
+    from botocore.customizations.sts import (
+        sts_default_setting as _sts_default_setting,
+    )
+except ImportError:
+    _sts_default_setting = 'legacy'
+
+_STS_DEFAULT_SETTINGS = {
+    'sts_regional_endpoints': (
+        'sts_regional_endpoints',
+        'AWS_STS_REGIONAL_ENDPOINTS',
+        _sts_default_setting,
+        None,
+    ),
+}
+BOTOCORE_DEFAUT_SESSION_VARIABLES.update(_STS_DEFAULT_SETTINGS)
+
+
+# A mapping for the s3 specific configuration vars. These are the configuration
+# vars that typically go in the s3 section of the config file. This mapping
+# follows the same schema as the previous session variable mapping.
+DEFAULT_S3_CONFIG_VARS = {
+    'addressing_style': (('s3', 'addressing_style'), None, None, None),
+    'use_accelerate_endpoint': (
+        ('s3', 'use_accelerate_endpoint'),
+        None,
+        None,
+        utils.ensure_boolean,
+    ),
+    'use_dualstack_endpoint': (
+        ('s3', 'use_dualstack_endpoint'),
+        None,
+        None,
+        utils.ensure_boolean,
+    ),
+    'payload_signing_enabled': (
+        ('s3', 'payload_signing_enabled'),
+        None,
+        None,
+        utils.ensure_boolean,
+    ),
+    'use_arn_region': (
+        ['s3_use_arn_region', ('s3', 'use_arn_region')],
+        'AWS_S3_USE_ARN_REGION',
+        None,
+        utils.ensure_boolean,
+    ),
+    'us_east_1_regional_endpoint': (
+        [
+            's3_us_east_1_regional_endpoint',
+            ('s3', 'us_east_1_regional_endpoint'),
+        ],
+        'AWS_S3_US_EAST_1_REGIONAL_ENDPOINT',
+        None,
+        None,
+    ),
+    's3_disable_multiregion_access_points': (
+        ('s3', 's3_disable_multiregion_access_points'),
+        'AWS_S3_DISABLE_MULTIREGION_ACCESS_POINTS',
+        None,
+        utils.ensure_boolean,
+    ),
+}
+# A mapping for the proxy specific configuration vars. These are
+# used to configure how botocore interacts with proxy setups while
+# sending requests.
+DEFAULT_PROXIES_CONFIG_VARS = {
+    'proxy_ca_bundle': ('proxy_ca_bundle', None, None, None),
+    'proxy_client_cert': ('proxy_client_cert', None, None, None),
+    'proxy_use_forwarding_for_https': (
+        'proxy_use_forwarding_for_https',
+        None,
+        None,
+        utils.normalize_boolean,
+    ),
+}
+
+
+def create_botocore_default_config_mapping(session):
+    chain_builder = ConfigChainFactory(session=session)
+    config_mapping = _create_config_chain_mapping(
+        chain_builder, BOTOCORE_DEFAUT_SESSION_VARIABLES
+    )
+    config_mapping['s3'] = SectionConfigProvider(
+        's3',
+        session,
+        _create_config_chain_mapping(chain_builder, DEFAULT_S3_CONFIG_VARS),
+    )
+    config_mapping['proxies_config'] = SectionConfigProvider(
+        'proxies_config',
+        session,
+        _create_config_chain_mapping(
+            chain_builder, DEFAULT_PROXIES_CONFIG_VARS
+        ),
+    )
+    return config_mapping
+
+
+def _create_config_chain_mapping(chain_builder, config_variables):
+    mapping = {}
+    for logical_name, config in config_variables.items():
+        mapping[logical_name] = chain_builder.create_config_chain(
+            instance_name=logical_name,
+            env_var_names=config[1],
+            config_property_names=config[0],
+            default=config[2],
+            conversion_func=config[3],
+        )
+    return mapping
+
+
+class DefaultConfigResolver:
+    def __init__(self, default_config_data):
+        self._base_default_config = default_config_data['base']
+        self._modes = default_config_data['modes']
+        self._resolved_default_configurations = {}
+
+    def _resolve_default_values_by_mode(self, mode):
+        default_config = self._base_default_config.copy()
+        modifications = self._modes.get(mode)
+
+        for config_var in modifications:
+            default_value = default_config[config_var]
+            modification_dict = modifications[config_var]
+            modification = list(modification_dict.keys())[0]
+            modification_value = modification_dict[modification]
+            if modification == 'multiply':
+                default_value *= modification_value
+            elif modification == 'add':
+                default_value += modification_value
+            elif modification == 'override':
+                default_value = modification_value
+            default_config[config_var] = default_value
+        return default_config
+
+    def get_default_modes(self):
+        default_modes = ['legacy', 'auto']
+        default_modes.extend(self._modes.keys())
+        return default_modes
+
+    def get_default_config_values(self, mode):
+        if mode not in self._resolved_default_configurations:
+            defaults = self._resolve_default_values_by_mode(mode)
+            self._resolved_default_configurations[mode] = defaults
+        return self._resolved_default_configurations[mode]
+
+
+class ConfigChainFactory:
+    """Factory class to create our most common configuration chain case.
+
+    This is a convenience class to construct configuration chains that follow
+    our most common pattern. This is to prevent ordering them incorrectly,
+    and to make the config chain construction more readable.
+    """
+
+    def __init__(self, session, environ=None):
+        """Initialize a ConfigChainFactory.
+
+        :type session: :class:`botocore.session.Session`
+        :param session: This is the session that should be used to look up
+            values from the config file.
+
+        :type environ: dict
+        :param environ: A mapping to use for environment variables. If this
+            is not provided it will default to use os.environ.
+        """
+        self._session = session
+        if environ is None:
+            environ = os.environ
+        self._environ = environ
+
+    def create_config_chain(
+        self,
+        instance_name=None,
+        env_var_names=None,
+        config_property_names=None,
+        default=None,
+        conversion_func=None,
+    ):
+        """Build a config chain following the standard botocore pattern.
+
+        In botocore most of our config chains follow the the precendence:
+        session_instance_variables, environment, config_file, default_value.
+
+        This is a convenience function for creating a chain that follow
+        that precendence.
+
+        :type instance_name: str
+        :param instance_name: This indicates what session instance variable
+            corresponds to this config value. If it is None it will not be
+            added to the chain.
+
+        :type env_var_names: str or list of str or None
+        :param env_var_names: One or more environment variable names to
+            search for this value. They are searched in order. If it is None
+            it will not be added to the chain.
+
+        :type config_property_names: str/tuple or list of str/tuple or None
+        :param config_property_names: One of more strings or tuples
+            representing the name of the key in the config file for this
+            config option. They are searched in order. If it is None it will
+            not be added to the chain.
+
+        :type default: Any
+        :param default: Any constant value to be returned.
+
+        :type conversion_func: None or callable
+        :param conversion_func: If this value is None then it has no effect on
+            the return type. Otherwise, it is treated as a function that will
+            conversion_func our provided type.
+
+        :rvalue: ConfigChain
+        :returns: A ConfigChain that resolves in the order env_var_names ->
+            config_property_name -> default. Any values that were none are
+            omitted form the chain.
+        """
+        providers = []
+        if instance_name is not None:
+            providers.append(
+                InstanceVarProvider(
+                    instance_var=instance_name, session=self._session
+                )
+            )
+        if env_var_names is not None:
+            providers.extend(self._get_env_providers(env_var_names))
+        if config_property_names is not None:
+            providers.extend(
+                self._get_scoped_config_providers(config_property_names)
+            )
+        if default is not None:
+            providers.append(ConstantProvider(value=default))
+
+        return ChainProvider(
+            providers=providers,
+            conversion_func=conversion_func,
+        )
+
+    def _get_env_providers(self, env_var_names):
+        env_var_providers = []
+        if not isinstance(env_var_names, list):
+            env_var_names = [env_var_names]
+        for env_var_name in env_var_names:
+            env_var_providers.append(
+                EnvironmentProvider(name=env_var_name, env=self._environ)
+            )
+        return env_var_providers
+
+    def _get_scoped_config_providers(self, config_property_names):
+        scoped_config_providers = []
+        if not isinstance(config_property_names, list):
+            config_property_names = [config_property_names]
+        for config_property_name in config_property_names:
+            scoped_config_providers.append(
+                ScopedConfigProvider(
+                    config_var_name=config_property_name,
+                    session=self._session,
+                )
+            )
+        return scoped_config_providers
+
+
+class ConfigValueStore:
+    """The ConfigValueStore object stores configuration values."""
+
+    def __init__(self, mapping=None):
+        """Initialize a ConfigValueStore.
+
+        :type mapping: dict
+        :param mapping: The mapping parameter is a map of string to a subclass
+            of BaseProvider. When a config variable is asked for via the
+            get_config_variable method, the corresponding provider will be
+            invoked to load the value.
+        """
+        self._overrides = {}
+        self._mapping = {}
+        if mapping is not None:
+            for logical_name, provider in mapping.items():
+                self.set_config_provider(logical_name, provider)
+
+    def __deepcopy__(self, memo):
+        config_store = ConfigValueStore(copy.deepcopy(self._mapping, memo))
+        for logical_name, override_value in self._overrides.items():
+            config_store.set_config_variable(logical_name, override_value)
+
+        return config_store
+
+    def __copy__(self):
+        config_store = ConfigValueStore(copy.copy(self._mapping))
+        for logical_name, override_value in self._overrides.items():
+            config_store.set_config_variable(logical_name, override_value)
+
+        return config_store
+
+    def get_config_variable(self, logical_name):
+        """
+        Retrieve the value associated with the specified logical_name
+        from the corresponding provider. If no value is found None will
+        be returned.
+
+        :type logical_name: str
+        :param logical_name: The logical name of the session variable
+            you want to retrieve.  This name will be mapped to the
+            appropriate environment variable name for this session as
+            well as the appropriate config file entry.
+
+        :returns: value of variable or None if not defined.
+        """
+        if logical_name in self._overrides:
+            return self._overrides[logical_name]
+        if logical_name not in self._mapping:
+            return None
+        provider = self._mapping[logical_name]
+        return provider.provide()
+
+    def get_config_provider(self, logical_name):
+        """
+        Retrieve the provider associated with the specified logical_name.
+        If no provider is found None will be returned.
+
+        :type logical_name: str
+        :param logical_name: The logical name of the session variable
+            you want to retrieve.  This name will be mapped to the
+            appropriate environment variable name for this session as
+            well as the appropriate config file entry.
+
+        :returns: configuration provider or None if not defined.
+        """
+        if (
+            logical_name in self._overrides
+            or logical_name not in self._mapping
+        ):
+            return None
+        provider = self._mapping[logical_name]
+        return provider
+
+    def set_config_variable(self, logical_name, value):
+        """Set a configuration variable to a specific value.
+
+        By using this method, you can override the normal lookup
+        process used in ``get_config_variable`` by explicitly setting
+        a value.  Subsequent calls to ``get_config_variable`` will
+        use the ``value``.  This gives you per-session specific
+        configuration values.
+
+        ::
+            >>> # Assume logical name 'foo' maps to env var 'FOO'
+            >>> os.environ['FOO'] = 'myvalue'
+            >>> s.get_config_variable('foo')
+            'myvalue'
+            >>> s.set_config_variable('foo', 'othervalue')
+            >>> s.get_config_variable('foo')
+            'othervalue'
+
+        :type logical_name: str
+        :param logical_name: The logical name of the session variable
+            you want to set.  These are the keys in ``SESSION_VARIABLES``.
+
+        :param value: The value to associate with the config variable.
+        """
+        self._overrides[logical_name] = value
+
+    def clear_config_variable(self, logical_name):
+        """Remove an override config variable from the session.
+
+        :type logical_name: str
+        :param logical_name: The name of the parameter to clear the override
+            value from.
+        """
+        self._overrides.pop(logical_name, None)
+
+    def set_config_provider(self, logical_name, provider):
+        """Set the provider for a config value.
+
+        This provides control over how a particular configuration value is
+        loaded. This replaces the provider for ``logical_name`` with the new
+        ``provider``.
+
+        :type logical_name: str
+        :param logical_name: The name of the config value to change the config
+            provider for.
+
+        :type provider: :class:`botocore.configprovider.BaseProvider`
+        :param provider: The new provider that should be responsible for
+            providing a value for the config named ``logical_name``.
+        """
+        self._mapping[logical_name] = provider
+
+
+class SmartDefaultsConfigStoreFactory:
+    def __init__(self, default_config_resolver, imds_region_provider):
+        self._default_config_resolver = default_config_resolver
+        self._imds_region_provider = imds_region_provider
+        # Initializing _instance_metadata_region as None so we
+        # can fetch region in a lazy fashion only when needed.
+        self._instance_metadata_region = None
+
+    def merge_smart_defaults(self, config_store, mode, region_name):
+        if mode == 'auto':
+            mode = self.resolve_auto_mode(region_name)
+        default_configs = (
+            self._default_config_resolver.get_default_config_values(mode)
+        )
+        for config_var in default_configs:
+            config_value = default_configs[config_var]
+            method = getattr(self, f'_set_{config_var}', None)
+            if method:
+                method(config_store, config_value)
+
+    def resolve_auto_mode(self, region_name):
+        current_region = None
+        if os.environ.get('AWS_EXECUTION_ENV'):
+            default_region = os.environ.get('AWS_DEFAULT_REGION')
+            current_region = os.environ.get('AWS_REGION', default_region)
+        if not current_region:
+            if self._instance_metadata_region:
+                current_region = self._instance_metadata_region
+            else:
+                try:
+                    current_region = self._imds_region_provider.provide()
+                    self._instance_metadata_region = current_region
+                except Exception:
+                    pass
+
+        if current_region:
+            if region_name == current_region:
+                return 'in-region'
+            else:
+                return 'cross-region'
+        return 'standard'
+
+    def _update_provider(self, config_store, variable, value):
+        original_provider = config_store.get_config_provider(variable)
+        default_provider = ConstantProvider(value)
+        if isinstance(original_provider, ChainProvider):
+            chain_provider_copy = copy.deepcopy(original_provider)
+            chain_provider_copy.set_default_provider(default_provider)
+            default_provider = chain_provider_copy
+        elif isinstance(original_provider, BaseProvider):
+            default_provider = ChainProvider(
+                providers=[original_provider, default_provider]
+            )
+        config_store.set_config_provider(variable, default_provider)
+
+    def _update_section_provider(
+        self, config_store, section_name, variable, value
+    ):
+        section_provider_copy = copy.deepcopy(
+            config_store.get_config_provider(section_name)
+        )
+        section_provider_copy.set_default_provider(
+            variable, ConstantProvider(value)
+        )
+        config_store.set_config_provider(section_name, section_provider_copy)
+
+    def _set_retryMode(self, config_store, value):
+        self._update_provider(config_store, 'retry_mode', value)
+
+    def _set_stsRegionalEndpoints(self, config_store, value):
+        self._update_provider(config_store, 'sts_regional_endpoints', value)
+
+    def _set_s3UsEast1RegionalEndpoints(self, config_store, value):
+        self._update_section_provider(
+            config_store, 's3', 'us_east_1_regional_endpoint', value
+        )
+
+    def _set_connectTimeoutInMillis(self, config_store, value):
+        self._update_provider(config_store, 'connect_timeout', value / 1000)
+
+
+class BaseProvider:
+    """Base class for configuration value providers.
+
+    A configuration provider has some method of providing a configuration
+    value.
+    """
+
+    def provide(self):
+        """Provide a config value."""
+        raise NotImplementedError('provide')
+
+
+class ChainProvider(BaseProvider):
+    """This provider wraps one or more other providers.
+
+    Each provider in the chain is called, the first one returning a non-None
+    value is then returned.
+    """
+
+    def __init__(self, providers=None, conversion_func=None):
+        """Initalize a ChainProvider.
+
+        :type providers: list
+        :param providers: The initial list of providers to check for values
+            when invoked.
+
+        :type conversion_func: None or callable
+        :param conversion_func: If this value is None then it has no affect on
+            the return type. Otherwise, it is treated as a function that will
+            transform provided value.
+        """
+        if providers is None:
+            providers = []
+        self._providers = providers
+        self._conversion_func = conversion_func
+
+    def __deepcopy__(self, memo):
+        return ChainProvider(
+            copy.deepcopy(self._providers, memo), self._conversion_func
+        )
+
+    def provide(self):
+        """Provide the value from the first provider to return non-None.
+
+        Each provider in the chain has its provide method called. The first
+        one in the chain to return a non-None value is the returned from the
+        ChainProvider. When no non-None value is found, None is returned.
+        """
+        for provider in self._providers:
+            value = provider.provide()
+            if value is not None:
+                return self._convert_type(value)
+        return None
+
+    def set_default_provider(self, default_provider):
+        if self._providers and isinstance(
+            self._providers[-1], ConstantProvider
+        ):
+            self._providers[-1] = default_provider
+        else:
+            self._providers.append(default_provider)
+
+        num_of_constants = sum(
+            isinstance(provider, ConstantProvider)
+            for provider in self._providers
+        )
+        if num_of_constants > 1:
+            logger.info(
+                'ChainProvider object contains multiple '
+                'instances of ConstantProvider objects'
+            )
+
+    def _convert_type(self, value):
+        if self._conversion_func is not None:
+            return self._conversion_func(value)
+        return value
+
+    def __repr__(self):
+        return '[{}]'.format(', '.join([str(p) for p in self._providers]))
+
+
+class InstanceVarProvider(BaseProvider):
+    """This class loads config values from the session instance vars."""
+
+    def __init__(self, instance_var, session):
+        """Initialize InstanceVarProvider.
+
+        :type instance_var: str
+        :param instance_var: The instance variable to load from the session.
+
+        :type session: :class:`botocore.session.Session`
+        :param session: The botocore session to get the loaded configuration
+            file variables from.
+        """
+        self._instance_var = instance_var
+        self._session = session
+
+    def __deepcopy__(self, memo):
+        return InstanceVarProvider(
+            copy.deepcopy(self._instance_var, memo), self._session
+        )
+
+    def provide(self):
+        """Provide a config value from the session instance vars."""
+        instance_vars = self._session.instance_variables()
+        value = instance_vars.get(self._instance_var)
+        return value
+
+    def __repr__(self):
+        return f'InstanceVarProvider(instance_var={self._instance_var}, session={self._session})'
+
+
+class ScopedConfigProvider(BaseProvider):
+    def __init__(self, config_var_name, session):
+        """Initialize ScopedConfigProvider.
+
+        :type config_var_name: str or tuple
+        :param config_var_name: The name of the config variable to load from
+            the configuration file. If the value is a tuple, it must only
+            consist of two items, where the first item represents the section
+            and the second item represents the config var name in the section.
+
+        :type session: :class:`botocore.session.Session`
+        :param session: The botocore session to get the loaded configuration
+            file variables from.
+        """
+        self._config_var_name = config_var_name
+        self._session = session
+
+    def __deepcopy__(self, memo):
+        return ScopedConfigProvider(
+            copy.deepcopy(self._config_var_name, memo), self._session
+        )
+
+    def provide(self):
+        """Provide a value from a config file property."""
+        scoped_config = self._session.get_scoped_config()
+        if isinstance(self._config_var_name, tuple):
+            section_config = scoped_config.get(self._config_var_name[0])
+            if not isinstance(section_config, dict):
+                return None
+            return section_config.get(self._config_var_name[1])
+        return scoped_config.get(self._config_var_name)
+
+    def __repr__(self):
+        return f'ScopedConfigProvider(config_var_name={self._config_var_name}, session={self._session})'
+
+
+class EnvironmentProvider(BaseProvider):
+    """This class loads config values from environment variables."""
+
+    def __init__(self, name, env):
+        """Initialize with the keys in the dictionary to check.
+
+        :type name: str
+        :param name: The key with that name will be loaded and returned.
+
+        :type env: dict
+        :param env: Environment variables dictionary to get variables from.
+        """
+        self._name = name
+        self._env = env
+
+    def __deepcopy__(self, memo):
+        return EnvironmentProvider(
+            copy.deepcopy(self._name, memo), copy.deepcopy(self._env, memo)
+        )
+
+    def provide(self):
+        """Provide a config value from a source dictionary."""
+        if self._name in self._env:
+            return self._env[self._name]
+        return None
+
+    def __repr__(self):
+        return f'EnvironmentProvider(name={self._name}, env={self._env})'
+
+
+class SectionConfigProvider(BaseProvider):
+    """Provides a dictionary from a section in the scoped config
+
+    This is useful for retrieving scoped config variables (i.e. s3) that have
+    their own set of config variables and resolving logic.
+    """
+
+    def __init__(self, section_name, session, override_providers=None):
+        self._section_name = section_name
+        self._session = session
+        self._scoped_config_provider = ScopedConfigProvider(
+            self._section_name, self._session
+        )
+        self._override_providers = override_providers
+        if self._override_providers is None:
+            self._override_providers = {}
+
+    def __deepcopy__(self, memo):
+        return SectionConfigProvider(
+            copy.deepcopy(self._section_name, memo),
+            self._session,
+            copy.deepcopy(self._override_providers, memo),
+        )
+
+    def provide(self):
+        section_config = self._scoped_config_provider.provide()
+        if section_config and not isinstance(section_config, dict):
+            logger.debug(
+                "The %s config key is not a dictionary type, "
+                "ignoring its value of: %s",
+                self._section_name,
+                section_config,
+            )
+            return None
+        for section_config_var, provider in self._override_providers.items():
+            provider_val = provider.provide()
+            if provider_val is not None:
+                if section_config is None:
+                    section_config = {}
+                section_config[section_config_var] = provider_val
+        return section_config
+
+    def set_default_provider(self, key, default_provider):
+        provider = self._override_providers.get(key)
+        if isinstance(provider, ChainProvider):
+            provider.set_default_provider(default_provider)
+            return
+        elif isinstance(provider, BaseProvider):
+            default_provider = ChainProvider(
+                providers=[provider, default_provider]
+            )
+        self._override_providers[key] = default_provider
+
+    def __repr__(self):
+        return (
+            f'SectionConfigProvider(section_name={self._section_name}, '
+            f'session={self._session}, '
+            f'override_providers={self._override_providers})'
+        )
+
+
+class ConstantProvider(BaseProvider):
+    """This provider provides a constant value."""
+
+    def __init__(self, value):
+        self._value = value
+
+    def __deepcopy__(self, memo):
+        return ConstantProvider(copy.deepcopy(self._value, memo))
+
+    def provide(self):
+        """Provide the constant value given during initialization."""
+        return self._value
+
+    def __repr__(self):
+        return f'ConstantProvider(value={self._value})'
+
+
+class ConfiguredEndpointProvider(BaseProvider):
+    """Lookup an endpoint URL from environment variable or shared config file.
+
+    NOTE: This class is considered private and is subject to abrupt breaking
+    changes or removal without prior announcement. Please do not use it
+    directly.
+    """
+
+    _ENDPOINT_URL_LOOKUP_ORDER = [
+        'environment_service',
+        'environment_global',
+        'config_service',
+        'config_global',
+    ]
+
+    def __init__(
+        self,
+        full_config,
+        scoped_config,
+        client_name,
+        environ=None,
+    ):
+        """Initialize a ConfiguredEndpointProviderChain.
+
+        :type full_config: dict
+        :param full_config: This is the dict representing the full
+            configuration file.
+
+        :type scoped_config: dict
+        :param scoped_config: This is the dict representing the configuration
+            for the current profile for the session.
+
+        :type client_name: str
+        :param client_name: The name used to instantiate a client using
+            botocore.session.Session.create_client.
+
+        :type environ: dict
+        :param environ: A mapping to use for environment variables. If this
+            is not provided it will default to use os.environ.
+        """
+        self._full_config = full_config
+        self._scoped_config = scoped_config
+        self._client_name = client_name
+        self._transformed_service_id = self._get_snake_case_service_id(
+            self._client_name
+        )
+        if environ is None:
+            environ = os.environ
+        self._environ = environ
+
+    def provide(self):
+        """Lookup the configured endpoint URL.
+
+        The order is:
+
+        1. The value provided by a service-specific environment variable.
+        2. The value provided by the global endpoint environment variable
+           (AWS_ENDPOINT_URL).
+        3. The value provided by a service-specific parameter from a services
+           definition section in the shared configuration file.
+        4. The value provided by the global parameter from a services
+           definition section in the shared configuration file.
+        """
+        for location in self._ENDPOINT_URL_LOOKUP_ORDER:
+            logger.debug(
+                'Looking for endpoint for %s via: %s',
+                self._client_name,
+                location,
+            )
+
+            endpoint_url = getattr(self, f'_get_endpoint_url_{location}')()
+
+            if endpoint_url:
+                logger.info(
+                    'Found endpoint for %s via: %s.',
+                    self._client_name,
+                    location,
+                )
+                return endpoint_url
+
+        logger.debug('No configured endpoint found.')
+        return None
+
+    def _get_snake_case_service_id(self, client_name):
+        # Get the service ID without loading the service data file, accounting
+        # for any aliases and standardizing the names with hyphens.
+        client_name = utils.SERVICE_NAME_ALIASES.get(client_name, client_name)
+        hyphenized_service_id = (
+            utils.CLIENT_NAME_TO_HYPHENIZED_SERVICE_ID_OVERRIDES.get(
+                client_name, client_name
+            )
+        )
+        return hyphenized_service_id.replace('-', '_')
+
+    def _get_service_env_var_name(self):
+        transformed_service_id_env = self._transformed_service_id.upper()
+        return f'AWS_ENDPOINT_URL_{transformed_service_id_env}'
+
+    def _get_services_config(self):
+        if 'services' not in self._scoped_config:
+            return {}
+
+        section_name = self._scoped_config['services']
+        services_section = self._full_config.get('services', {}).get(
+            section_name
+        )
+
+        if not services_section:
+            error_msg = (
+                f'The profile is configured to use the services '
+                f'section but the "{section_name}" services '
+                f'configuration does not exist.'
+            )
+            raise InvalidConfigError(error_msg=error_msg)
+
+        return services_section
+
+    def _get_endpoint_url_config_service(self):
+        snakecase_service_id = self._transformed_service_id.lower()
+        return (
+            self._get_services_config()
+            .get(snakecase_service_id, {})
+            .get('endpoint_url')
+        )
+
+    def _get_endpoint_url_config_global(self):
+        return self._scoped_config.get('endpoint_url')
+
+    def _get_endpoint_url_environment_service(self):
+        return EnvironmentProvider(
+            name=self._get_service_env_var_name(), env=self._environ
+        ).provide()
+
+    def _get_endpoint_url_environment_global(self):
+        return EnvironmentProvider(
+            name='AWS_ENDPOINT_URL', env=self._environ
+        ).provide()