Implement configured endpoint URL resolution

Adds a config provider to resolve the endpoint URL provided in an
environment variable or shared configuration file. It resolves the
endpoint in the following manner:

1. The value provided through the `endpoint_url` parameter provided to
   the client constructor.
2. The value provided by a service-specific environment variable.
3. The value provided by the global endpoint environment variable
   (`AWS_ENDPOINT_URL`).
4. The value provided by a service-specific parameter from a services
   definition section in the shared configuration file.
5. The value provided by the global parameter from a services definition
   section in the shared configuration file.
6. The value resolved when no configured endpoint URL is provided.

The endpoint config provider uses the client name (name used to
instantiate a client object) for construction and add to the config
value store. This uses multiple lookups to handle service name changes
for backwards compatibility.

.changes/next-release/feature-configuration-3829.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "configuration",
+  "description": "Configure the endpoint URL in the shared configuration file or via an environment variable for a specific AWS service or all AWS services."
+}

botocore/args.py

@@ -107,7 +107,7 @@ def get_client_args(
         s3_config = final_args['s3_config']
         partition = endpoint_config['metadata'].get('partition', None)
         socket_options = final_args['socket_options']
-
+        configured_endpoint_url = final_args['configured_endpoint_url']
         signing_region = endpoint_config['signing_region']
         endpoint_region_name = endpoint_config['region_name']
 
@@ -152,7 +152,7 @@ def get_client_args(
             service_model,
             endpoint_region_name,
             region_name,
-            endpoint_url,
+            configured_endpoint_url,
             endpoint,
             is_secure,
             endpoint_bridge,
@@ -202,10 +202,16 @@ def compute_client_args(
                 user_agent += ' %s' % client_config.user_agent_extra
 
         s3_config = self.compute_s3_config(client_config)
+
+        configured_endpoint_url = self._compute_configured_endpoint_url(
+            client_config=client_config,
+            endpoint_url=endpoint_url,
+        )
+
         endpoint_config = self._compute_endpoint_config(
             service_name=service_name,
             region_name=region_name,
-            endpoint_url=endpoint_url,
+            endpoint_url=configured_endpoint_url,
             is_secure=is_secure,
             endpoint_bridge=endpoint_bridge,
             s3_config=s3_config,
@@ -250,6 +256,7 @@ def compute_client_args(
             'service_name': service_name,
             'parameter_validation': parameter_validation,
             'user_agent': user_agent,
+            'configured_endpoint_url': configured_endpoint_url,
             'endpoint_config': endpoint_config,
             'protocol': protocol,
             'config_kwargs': config_kwargs,
@@ -259,6 +266,25 @@ def compute_client_args(
             ),
         }
 
+    def _compute_configured_endpoint_url(self, client_config, endpoint_url):
+        if endpoint_url is not None or self._ignore_configured_endpoint_urls(
+            client_config
+        ):
+            return endpoint_url
+
+        return self._config_store.get_config_variable('endpoint_url')
+
+    def _ignore_configured_endpoint_urls(self, client_config):
+        if (
+            client_config
+            and client_config.ignore_configured_endpoint_urls is not None
+        ):
+            return client_config.ignore_configured_endpoint_urls
+
+        return self._config_store.get_config_variable(
+            'ignore_configured_endpoint_urls'
+        )
+
     def compute_s3_config(self, client_config):
         s3_configuration = self._config_store.get_config_variable('s3')
 

botocore/config.py

@@ -188,6 +188,13 @@ class Config:
 
         Defaults to None.
 
+    :type ignore_configured_endpoint_urls: bool
+    :param ignore_configured_endpoint_urls: Setting to True disables use
+        of endpoint URLs provided via environment variables and
+        the shared configuration file.
+
+        Defaults to None.
+
     :type tcp_keepalive: bool
     :param tcp_keepalive: Enables the TCP Keep-Alive socket option used when
         creating new connections if set to True.
@@ -214,6 +221,7 @@ class Config:
             ('endpoint_discovery_enabled', None),
             ('use_dualstack_endpoint', None),
             ('use_fips_endpoint', None),
+            ('ignore_configured_endpoint_urls', None),
             ('defaults_mode', None),
             ('tcp_keepalive', None),
         ]

botocore/configloader.py

@@ -200,6 +200,17 @@ def _parse_nested(config_value):
     return parsed
 
 
+def _parse_section(key, values):
+    result = {}
+    try:
+        parts = shlex.split(key)
+    except ValueError:
+        return result
+    if len(parts) == 2:
+        result[parts[1]] = values
+    return result
+
+
 def build_profile_map(parsed_ini_config):
     """Convert the parsed INI config into a profile map.
 
@@ -254,22 +265,15 @@ def build_profile_map(parsed_ini_config):
     parsed_config = copy.deepcopy(parsed_ini_config)
     profiles = {}
     sso_sessions = {}
+    services = {}
     final_config = {}
     for key, values in parsed_config.items():
         if key.startswith("profile"):
-            try:
-                parts = shlex.split(key)
-            except ValueError:
-                continue
-            if len(parts) == 2:
-                profiles[parts[1]] = values
+            profiles.update(_parse_section(key, values))
         elif key.startswith("sso-session"):
-            try:
-                parts = shlex.split(key)
-            except ValueError:
-                continue
-            if len(parts) == 2:
-                sso_sessions[parts[1]] = values
+            sso_sessions.update(_parse_section(key, values))
+        elif key.startswith("services"):
+            services.update(_parse_section(key, values))
         elif key == 'default':
             # default section is special and is considered a profile
             # name but we don't require you use 'profile "default"'
@@ -279,4 +283,5 @@ def build_profile_map(parsed_ini_config):
             final_config[key] = values
     final_config['profiles'] = profiles
     final_config['sso_sessions'] = sso_sessions
+    final_config['services'] = services
     return final_config

botocore/configprovider.py

@@ -18,6 +18,7 @@
 import os
 
 from botocore import utils
+from botocore.exceptions import InvalidConfigError
 
 logger = logging.getLogger(__name__)
 
@@ -108,6 +109,12 @@
         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.
@@ -851,3 +858,159 @@ def provide(self):
 
     def __repr__(self):
         return 'ConstantProvider(value=%s)' % self._value
+
+
+class ConfiguredEndpointProvider(BaseProvider):
+    """Lookup an endpoint URL from environment variable or shared cong 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
+
+        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):
+        # Use lookups to get the serice ID without loading the service data
+        # file. `client_name`` refers to the name used to instantiate a client
+        # through botocore.session.Session.create_client (parameter name there
+        # is `service_name`). This client name is generally the hyphenized
+        # service ID.  We need to look up services that have been renamed
+        # (SERVICE_NAME_ALIASES) as well as services that do not use
+        # the service ID as their data directory name
+        # (CLIENT_NAME_TO_HYPHENIZED_SERVICE_ID_OVERRIDES).
+        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._get_snake_case_service_id(
+            self._client_name
+        ).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._get_snake_case_service_id(
+            self._client_name
+        ).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()
+
+    def __repr__(self):
+        return (
+            f'ConfiguredEndpointProvider('
+            f'full_config={self._full_config}, '
+            f'scoped_config={self._scoped_config}, '
+            f'client_name={self._client_name}, '
+            f'environ={self._environ})'
+        )

botocore/handlers.py

@@ -73,6 +73,7 @@
 from botocore.exceptions import MissingServiceIdError  # noqa
 from botocore.utils import hyphenize_service_id  # noqa
 from botocore.utils import is_global_accesspoint  # noqa
+from botocore.utils import SERVICE_NAME_ALIASES  # noqa
 
 
 logger = logging.getLogger(__name__)
@@ -99,8 +100,6 @@
 S3_SIGNING_NAMES = ('s3', 's3-outposts', 's3-object-lambda')
 VERSION_ID_SUFFIX = re.compile(r'\?versionId=[^\s]+$')
 
-SERVICE_NAME_ALIASES = {'runtime.sagemaker': 'sagemaker-runtime'}
-
 
 def handle_service_name_alias(service_name, **kwargs):
     return SERVICE_NAME_ALIASES.get(service_name, service_name)

botocore/session.py

@@ -42,6 +42,7 @@
 from botocore.configprovider import (
     BOTOCORE_DEFAUT_SESSION_VARIABLES,
     ConfigChainFactory,
+    ConfiguredEndpointProvider,
     ConfigValueStore,
     DefaultConfigResolver,
     SmartDefaultsConfigStoreFactory,
@@ -962,6 +963,12 @@ def create_client(
             smart_defaults_factory.merge_smart_defaults(
                 config_store, defaults_mode, region_name
             )
+
+        self._add_configured_endpoint_provider(
+            client_name=service_name,
+            config_store=config_store,
+        )
+
         client_creator = botocore.client.ClientCreator(
             loader,
             endpoint_resolver,
@@ -1030,6 +1037,17 @@ def _resolve_defaults_mode(self, client_config, config_store):
 
         return lmode
 
+    def _add_configured_endpoint_provider(self, client_name, config_store):
+        chain = ConfiguredEndpointProvider(
+            full_config=self.full_config,
+            scoped_config=self.get_scoped_config(),
+            client_name=client_name,
+        )
+        config_store.set_config_provider(
+            logical_name='endpoint_url',
+            provider=chain,
+        )
+
     def _missing_cred_vars(self, access_key, secret_key):
         if access_key is not None and secret_key is None:
             return 'aws_secret_access_key'