FM / botocore

/opt/imh-python/lib/python3.9/site-packages/botocore/

/opt/imh-python/lib/python3.9/site-packages/botocore

Name	Size	Mode	Actions
data/	-	0755	rm
docs/	-	0755	rm
retries/	-	0755	rm
vendored/	-	0755	rm
__pycache__/	-	0755	rm
args.py	16199	0644	edit dl rm
auth.py	34966	0644	edit dl rm
awsrequest.py	23247	0644	edit dl rm
cacert.pem	271088	0644	edit dl rm
client.py	41514	0644	edit dl rm
compat.py	11715	0644	edit dl rm
config.py	12721	0644	edit dl rm
configloader.py	9580	0644	edit dl rm
configprovider.py	22235	0644	edit dl rm
credentials.py	81582	0644	edit dl rm
discovery.py	11031	0644	edit dl rm
endpoint.py	13811	0644	edit dl rm
errorfactory.py	3727	0644	edit dl rm
eventstream.py	20517	0644	edit dl rm
exceptions.py	19927	0644	edit dl rm
handlers.py	45142	0644	edit dl rm
history.py	1748	0644	edit dl rm
hooks.py	24573	0644	edit dl rm
httpsession.py	13547	0644	edit dl rm
loaders.py	17355	0644	edit dl rm
model.py	28352	0644	edit dl rm
monitoring.py	20586	0644	edit dl rm
paginate.py	27128	0644	edit dl rm
parsers.py	42301	0644	edit dl rm
regions.py	8975	0644	edit dl rm
response.py	6434	0644	edit dl rm
retryhandler.py	13781	0644	edit dl rm
serialize.py	30430	0644	edit dl rm
session.py	43433	0644	edit dl rm
signers.py	28234	0644	edit dl rm
stub.py	14361	0644	edit dl rm
translate.py	3412	0644	edit dl rm
utils.py	86492	0644	edit dl rm
validate.py	11323	0644	edit dl rm
waiter.py	13860	0644	edit dl rm
__init__.py	3881	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/botocore/args.py (16199B)

# Copyright 2016 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.
"""Internal module to help with normalizing botocore client args.

This module (and all function/classes within this module) should be
considered internal, and *not* a public API.

"""
import copy
import logging
import socket

import botocore.exceptions
import botocore.serialize
import botocore.utils
from botocore.signers import RequestSigner
from botocore.config import Config
from botocore.endpoint import EndpointCreator

logger = logging.getLogger(__name__)

VALID_REGIONAL_ENDPOINTS_CONFIG = [
    'legacy',
    'regional',
]
LEGACY_GLOBAL_STS_REGIONS = [
    'ap-northeast-1',
    'ap-south-1',
    'ap-southeast-1',
    'ap-southeast-2',
    'aws-global',
    'ca-central-1',
    'eu-central-1',
    'eu-north-1',
    'eu-west-1',
    'eu-west-2',
    'eu-west-3',
    'sa-east-1',
    'us-east-1',
    'us-east-2',
    'us-west-1',
    'us-west-2',
]

class ClientArgsCreator(object):
    def __init__(self, event_emitter, user_agent, response_parser_factory,
                 loader, exceptions_factory, config_store):
        self._event_emitter = event_emitter
        self._user_agent = user_agent
        self._response_parser_factory = response_parser_factory
        self._loader = loader
        self._exceptions_factory = exceptions_factory
        self._config_store = config_store

def get_client_args(self, service_model, region_name, is_secure,
                        endpoint_url, verify, credentials, scoped_config,
                        client_config, endpoint_bridge):
        final_args = self.compute_client_args(
            service_model, client_config, endpoint_bridge, region_name,
            endpoint_url, is_secure, scoped_config)

service_name = final_args['service_name']
        parameter_validation = final_args['parameter_validation']
        endpoint_config = final_args['endpoint_config']
        protocol = final_args['protocol']
        config_kwargs = final_args['config_kwargs']
        s3_config = final_args['s3_config']
        partition = endpoint_config['metadata'].get('partition', None)
        socket_options = final_args['socket_options']

signing_region = endpoint_config['signing_region']
        endpoint_region_name = endpoint_config['region_name']

event_emitter = copy.copy(self._event_emitter)
        signer = RequestSigner(
            service_model.service_id, signing_region,
            endpoint_config['signing_name'],
            endpoint_config['signature_version'],
            credentials, event_emitter
        )

config_kwargs['s3'] = s3_config
        new_config = Config(**config_kwargs)
        endpoint_creator = EndpointCreator(event_emitter)

endpoint = endpoint_creator.create_endpoint(
            service_model, region_name=endpoint_region_name,
            endpoint_url=endpoint_config['endpoint_url'], verify=verify,
            response_parser_factory=self._response_parser_factory,
            max_pool_connections=new_config.max_pool_connections,
            proxies=new_config.proxies,
            timeout=(new_config.connect_timeout, new_config.read_timeout),
            socket_options=socket_options,
            client_cert=new_config.client_cert,
            proxies_config=new_config.proxies_config)

serializer = botocore.serialize.create_serializer(
            protocol, parameter_validation)
        response_parser = botocore.parsers.create_parser(protocol)
        return {
            'serializer': serializer,
            'endpoint': endpoint,
            'response_parser': response_parser,
            'event_emitter': event_emitter,
            'request_signer': signer,
            'service_model': service_model,
            'loader': self._loader,
            'client_config': new_config,
            'partition': partition,
            'exceptions_factory': self._exceptions_factory
        }

def compute_client_args(self, service_model, client_config,
                            endpoint_bridge, region_name, endpoint_url,
                            is_secure, scoped_config):
        service_name = service_model.endpoint_prefix
        protocol = service_model.metadata['protocol']
        parameter_validation = True
        if client_config and not client_config.parameter_validation:
            parameter_validation = False
        elif scoped_config:
            raw_value = scoped_config.get('parameter_validation')
            if raw_value is not None:
                parameter_validation = botocore.utils.ensure_boolean(raw_value)

# Override the user agent if specified in the client config.
        user_agent = self._user_agent
        if client_config is not None:
            if client_config.user_agent is not None:
                user_agent = client_config.user_agent
            if client_config.user_agent_extra is not None:
                user_agent += ' %s' % client_config.user_agent_extra

s3_config = self.compute_s3_config(client_config)
        endpoint_config = self._compute_endpoint_config(
            service_name=service_name,
            region_name=region_name,
            endpoint_url=endpoint_url,
            is_secure=is_secure,
            endpoint_bridge=endpoint_bridge,
            s3_config=s3_config,
        )
        # Create a new client config to be passed to the client based
        # on the final values. We do not want the user to be able
        # to try to modify an existing client with a client config.
        config_kwargs = dict(
            region_name=endpoint_config['region_name'],
            signature_version=endpoint_config['signature_version'],
            user_agent=user_agent)
        if client_config is not None:
            config_kwargs.update(
                connect_timeout=client_config.connect_timeout,
                read_timeout=client_config.read_timeout,
                max_pool_connections=client_config.max_pool_connections,
                proxies=client_config.proxies,
                proxies_config=client_config.proxies_config,
                retries=client_config.retries,
                client_cert=client_config.client_cert,
                inject_host_prefix=client_config.inject_host_prefix,
            )
        self._compute_retry_config(config_kwargs)
        s3_config = self.compute_s3_config(client_config)
        return {
            'service_name': service_name,
            'parameter_validation': parameter_validation,
            'user_agent': user_agent,
            'endpoint_config': endpoint_config,
            'protocol': protocol,
            'config_kwargs': config_kwargs,
            's3_config': s3_config,
            'socket_options': self._compute_socket_options(scoped_config)
        }

def compute_s3_config(self, client_config):
        s3_configuration = self._config_store.get_config_variable('s3')

# Next specific client config values takes precedence over
        # specific values in the scoped config.
        if client_config is not None:
            if client_config.s3 is not None:
                if s3_configuration is None:
                    s3_configuration = client_config.s3
                else:
                    # The current s3_configuration dictionary may be
                    # from a source that only should be read from so
                    # we want to be safe and just make a copy of it to modify
                    # before it actually gets updated.
                    s3_configuration = s3_configuration.copy()
                    s3_configuration.update(client_config.s3)

return s3_configuration

def _compute_endpoint_config(self, service_name, region_name, endpoint_url,
                                 is_secure, endpoint_bridge, s3_config):
        resolve_endpoint_kwargs = {
            'service_name': service_name,
            'region_name': region_name,
            'endpoint_url': endpoint_url,
            'is_secure': is_secure,
            'endpoint_bridge': endpoint_bridge,
        }
        if service_name == 's3':
            return self._compute_s3_endpoint_config(
                s3_config=s3_config, **resolve_endpoint_kwargs)
        if service_name == 'sts':
            return self._compute_sts_endpoint_config(**resolve_endpoint_kwargs)
        return self._resolve_endpoint(**resolve_endpoint_kwargs)

def _compute_s3_endpoint_config(self, s3_config,
                                    **resolve_endpoint_kwargs):
        force_s3_global = self._should_force_s3_global(
            resolve_endpoint_kwargs['region_name'], s3_config)
        if force_s3_global:
            resolve_endpoint_kwargs['region_name'] = None
        endpoint_config = self._resolve_endpoint(**resolve_endpoint_kwargs)
        self._set_region_if_custom_s3_endpoint(
            endpoint_config, resolve_endpoint_kwargs['endpoint_bridge'])
        # For backwards compatibility reasons, we want to make sure the
        # client.meta.region_name will remain us-east-1 if we forced the
        # endpoint to be the global region. Specifically, if this value
        # changes to aws-global, it breaks logic where a user is checking
        # for us-east-1 as the global endpoint such as in creating buckets.
        if force_s3_global and endpoint_config['region_name'] == 'aws-global':
            endpoint_config['region_name'] = 'us-east-1'
        return endpoint_config

def _should_force_s3_global(self, region_name, s3_config):
        s3_regional_config = 'legacy'
        if s3_config and 'us_east_1_regional_endpoint' in s3_config:
            s3_regional_config = s3_config['us_east_1_regional_endpoint']
            self._validate_s3_regional_config(s3_regional_config)
        return (
            s3_regional_config == 'legacy' and
            region_name in ['us-east-1', None]
        )

def _validate_s3_regional_config(self, config_val):
        if config_val not in VALID_REGIONAL_ENDPOINTS_CONFIG:
            raise botocore.exceptions.\
                InvalidS3UsEast1RegionalEndpointConfigError(
                    s3_us_east_1_regional_endpoint_config=config_val)

def _set_region_if_custom_s3_endpoint(self, endpoint_config,
                                          endpoint_bridge):
        # If a user is providing a custom URL, the endpoint resolver will
        # refuse to infer a signing region. If we want to default to s3v4,
        # we have to account for this.
        if endpoint_config['signing_region'] is None \
                and endpoint_config['region_name'] is None:
            endpoint = endpoint_bridge.resolve('s3')
            endpoint_config['signing_region'] = endpoint['signing_region']
            endpoint_config['region_name'] = endpoint['region_name']

def _compute_sts_endpoint_config(self, **resolve_endpoint_kwargs):
        endpoint_config = self._resolve_endpoint(**resolve_endpoint_kwargs)
        if self._should_set_global_sts_endpoint(
                resolve_endpoint_kwargs['region_name'],
                resolve_endpoint_kwargs['endpoint_url']):
            self._set_global_sts_endpoint(
                endpoint_config, resolve_endpoint_kwargs['is_secure'])
        return endpoint_config

def _should_set_global_sts_endpoint(self, region_name, endpoint_url):
        if endpoint_url:
            return False
        return (
            self._get_sts_regional_endpoints_config() == 'legacy' and
            region_name in LEGACY_GLOBAL_STS_REGIONS
        )

def _get_sts_regional_endpoints_config(self):
        sts_regional_endpoints_config = self._config_store.get_config_variable(
            'sts_regional_endpoints')
        if not sts_regional_endpoints_config:
            sts_regional_endpoints_config = 'legacy'
        if sts_regional_endpoints_config not in \
                VALID_REGIONAL_ENDPOINTS_CONFIG:
            raise botocore.exceptions.InvalidSTSRegionalEndpointsConfigError(
                sts_regional_endpoints_config=sts_regional_endpoints_config)
        return sts_regional_endpoints_config

def _set_global_sts_endpoint(self, endpoint_config, is_secure):
        scheme = 'https' if is_secure else 'http'
        endpoint_config['endpoint_url'] = '%s://sts.amazonaws.com' % scheme
        endpoint_config['signing_region'] = 'us-east-1'

def _resolve_endpoint(self, service_name, region_name,
                          endpoint_url, is_secure, endpoint_bridge):
        return endpoint_bridge.resolve(
            service_name, region_name, endpoint_url, is_secure)

def _compute_socket_options(self, scoped_config):
        # This disables Nagle's algorithm and is the default socket options
        # in urllib3.
        socket_options = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
        if scoped_config:
            # Enables TCP Keepalive if specified in shared config file.
            if self._ensure_boolean(scoped_config.get('tcp_keepalive', False)):
                socket_options.append(
                    (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1))
        return socket_options

def _compute_retry_config(self, config_kwargs):
        self._compute_retry_max_attempts(config_kwargs)
        self._compute_retry_mode(config_kwargs)

def _compute_retry_max_attempts(self, config_kwargs):
        # There's a pre-existing max_attempts client config value that actually
        # means max *retry* attempts.  There's also a `max_attempts` we pull
        # from the config store that means *total attempts*, which includes the
        # intitial request.  We can't change what `max_attempts` means in
        # client config so we try to normalize everything to a new
        # "total_max_attempts" variable.  We ensure that after this, the only
        # configuration for "max attempts" is the 'total_max_attempts' key.
        # An explicitly provided max_attempts in the client config
        # overrides everything.
        retries = config_kwargs.get('retries')
        if retries is not None:
            if 'total_max_attempts' in retries:
                retries.pop('max_attempts', None)
                return
            if 'max_attempts' in retries:
                value = retries.pop('max_attempts')
                # client config max_attempts means total retries so we
                # have to add one for 'total_max_attempts' to account
                # for the initial request.
                retries['total_max_attempts'] = value + 1
                return
        # Otherwise we'll check the config store which checks env vars,
        # config files, etc.  There is no default value for max_attempts
        # so if this returns None and we don't set a default value here.
        max_attempts = self._config_store.get_config_variable('max_attempts')
        if max_attempts is not None:
            if retries is None:
                retries = {}
                config_kwargs['retries'] = retries
            retries['total_max_attempts'] = max_attempts

def _compute_retry_mode(self, config_kwargs):
        retries = config_kwargs.get('retries')
        if retries is None:
            retries = {}
            config_kwargs['retries'] = retries
        elif 'mode' in retries:
            # If there's a retry mode explicitly set in the client config
            # that overrides everything.
            return
        retry_mode = self._config_store.get_config_variable('retry_mode')
        if retry_mode is None:
            retry_mode = 'legacy'
        retries['mode'] = retry_mode

def _ensure_boolean(self, val):
        if isinstance(val, bool):
            return val
        else:
            return val.lower() == 'true'

cmd: