123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327 |
- # Copyright (c) 2012-2013 Mitch Garnaat http://garnaat.org/
- # Copyright 2012-2014 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.
- import os
- import logging
- import time
- import threading
- from botocore.vendored import six
- from botocore.awsrequest import create_request_object
- from botocore.exceptions import HTTPClientError
- from botocore.httpsession import URLLib3Session
- from botocore.utils import is_valid_endpoint_url, get_environ_proxies
- from botocore.hooks import first_non_none_response
- from botocore.history import get_global_history_recorder
- from botocore.response import StreamingBody
- from botocore import parsers
- logger = logging.getLogger(__name__)
- history_recorder = get_global_history_recorder()
- DEFAULT_TIMEOUT = 60
- MAX_POOL_CONNECTIONS = 10
- def convert_to_response_dict(http_response, operation_model):
- """Convert an HTTP response object to a request dict.
- This converts the requests library's HTTP response object to
- a dictionary.
- :type http_response: botocore.vendored.requests.model.Response
- :param http_response: The HTTP response from an AWS service request.
- :rtype: dict
- :return: A response dictionary which will contain the following keys:
- * headers (dict)
- * status_code (int)
- * body (string or file-like object)
- """
- response_dict = {
- 'headers': http_response.headers,
- 'status_code': http_response.status_code,
- 'context': {
- 'operation_name': operation_model.name,
- }
- }
- if response_dict['status_code'] >= 300:
- response_dict['body'] = http_response.content
- elif operation_model.has_event_stream_output:
- response_dict['body'] = http_response.raw
- elif operation_model.has_streaming_output:
- length = response_dict['headers'].get('content-length')
- response_dict['body'] = StreamingBody(http_response.raw, length)
- else:
- response_dict['body'] = http_response.content
- return response_dict
- class Endpoint(object):
- """
- Represents an endpoint for a particular service in a specific
- region. Only an endpoint can make requests.
- :ivar service: The Service object that describes this endpoints
- service.
- :ivar host: The fully qualified endpoint hostname.
- :ivar session: The session object.
- """
- def __init__(self, host, endpoint_prefix, event_emitter,
- response_parser_factory=None, http_session=None):
- self._endpoint_prefix = endpoint_prefix
- self._event_emitter = event_emitter
- self.host = host
- self._lock = threading.Lock()
- if response_parser_factory is None:
- response_parser_factory = parsers.ResponseParserFactory()
- self._response_parser_factory = response_parser_factory
- self.http_session = http_session
- if self.http_session is None:
- self.http_session = URLLib3Session()
- def __repr__(self):
- return '%s(%s)' % (self._endpoint_prefix, self.host)
- def make_request(self, operation_model, request_dict):
- logger.debug("Making request for %s with params: %s",
- operation_model, request_dict)
- return self._send_request(request_dict, operation_model)
- def create_request(self, params, operation_model=None):
- request = create_request_object(params)
- if operation_model:
- request.stream_output = any([
- operation_model.has_streaming_output,
- operation_model.has_event_stream_output
- ])
- service_id = operation_model.service_model.service_id.hyphenize()
- event_name = 'request-created.{service_id}.{op_name}'.format(
- service_id=service_id,
- op_name=operation_model.name)
- self._event_emitter.emit(event_name, request=request,
- operation_name=operation_model.name)
- prepared_request = self.prepare_request(request)
- return prepared_request
- def _encode_headers(self, headers):
- # In place encoding of headers to utf-8 if they are unicode.
- for key, value in headers.items():
- if isinstance(value, six.text_type):
- headers[key] = value.encode('utf-8')
- def prepare_request(self, request):
- self._encode_headers(request.headers)
- return request.prepare()
- def _send_request(self, request_dict, operation_model):
- attempts = 1
- request = self.create_request(request_dict, operation_model)
- context = request_dict['context']
- success_response, exception = self._get_response(
- request, operation_model, context)
- while self._needs_retry(attempts, operation_model, request_dict,
- success_response, exception):
- attempts += 1
- # If there is a stream associated with the request, we need
- # to reset it before attempting to send the request again.
- # This will ensure that we resend the entire contents of the
- # body.
- request.reset_stream()
- # Create a new request when retried (including a new signature).
- request = self.create_request(
- request_dict, operation_model)
- success_response, exception = self._get_response(
- request, operation_model, context)
- if success_response is not None and \
- 'ResponseMetadata' in success_response[1]:
- # We want to share num retries, not num attempts.
- total_retries = attempts - 1
- success_response[1]['ResponseMetadata']['RetryAttempts'] = \
- total_retries
- if exception is not None:
- raise exception
- else:
- return success_response
- def _get_response(self, request, operation_model, context):
- # This will return a tuple of (success_response, exception)
- # and success_response is itself a tuple of
- # (http_response, parsed_dict).
- # If an exception occurs then the success_response is None.
- # If no exception occurs then exception is None.
- success_response, exception = self._do_get_response(
- request, operation_model)
- kwargs_to_emit = {
- 'response_dict': None,
- 'parsed_response': None,
- 'context': context,
- 'exception': exception,
- }
- if success_response is not None:
- http_response, parsed_response = success_response
- kwargs_to_emit['parsed_response'] = parsed_response
- kwargs_to_emit['response_dict'] = convert_to_response_dict(
- http_response, operation_model)
- service_id = operation_model.service_model.service_id.hyphenize()
- self._event_emitter.emit(
- 'response-received.%s.%s' % (
- service_id, operation_model.name), **kwargs_to_emit)
- return success_response, exception
- def _do_get_response(self, request, operation_model):
- try:
- logger.debug("Sending http request: %s", request)
- history_recorder.record('HTTP_REQUEST', {
- 'method': request.method,
- 'headers': request.headers,
- 'streaming': operation_model.has_streaming_input,
- 'url': request.url,
- 'body': request.body
- })
- service_id = operation_model.service_model.service_id.hyphenize()
- event_name = 'before-send.%s.%s' % (service_id, operation_model.name)
- responses = self._event_emitter.emit(event_name, request=request)
- http_response = first_non_none_response(responses)
- if http_response is None:
- http_response = self._send(request)
- except HTTPClientError as e:
- return (None, e)
- except Exception as e:
- logger.debug("Exception received when sending HTTP request.",
- exc_info=True)
- return (None, e)
- # This returns the http_response and the parsed_data.
- response_dict = convert_to_response_dict(http_response, operation_model)
- http_response_record_dict = response_dict.copy()
- http_response_record_dict['streaming'] = \
- operation_model.has_streaming_output
- history_recorder.record('HTTP_RESPONSE', http_response_record_dict)
- protocol = operation_model.metadata['protocol']
- parser = self._response_parser_factory.create_parser(protocol)
- parsed_response = parser.parse(
- response_dict, operation_model.output_shape)
- # Do a second parsing pass to pick up on any modeled error fields
- # NOTE: Ideally, we would push this down into the parser classes but
- # they currently have no reference to the operation or service model
- # The parsers should probably take the operation model instead of
- # output shape but we can't change that now
- if http_response.status_code >= 300:
- self._add_modeled_error_fields(
- response_dict, parsed_response,
- operation_model, parser,
- )
- history_recorder.record('PARSED_RESPONSE', parsed_response)
- return (http_response, parsed_response), None
- def _add_modeled_error_fields(
- self, response_dict, parsed_response,
- operation_model, parser,
- ):
- error_code = parsed_response.get("Error", {}).get("Code")
- if error_code is None:
- return
- service_model = operation_model.service_model
- error_shape = service_model.shape_for_error_code(error_code)
- if error_shape is None:
- return
- modeled_parse = parser.parse(response_dict, error_shape)
- # TODO: avoid naming conflicts with ResponseMetadata and Error
- parsed_response.update(modeled_parse)
- def _needs_retry(self, attempts, operation_model, request_dict,
- response=None, caught_exception=None):
- service_id = operation_model.service_model.service_id.hyphenize()
- event_name = 'needs-retry.%s.%s' % (
- service_id,
- operation_model.name)
- responses = self._event_emitter.emit(
- event_name, response=response, endpoint=self,
- operation=operation_model, attempts=attempts,
- caught_exception=caught_exception, request_dict=request_dict)
- handler_response = first_non_none_response(responses)
- if handler_response is None:
- return False
- else:
- # Request needs to be retried, and we need to sleep
- # for the specified number of times.
- logger.debug("Response received to retry, sleeping for "
- "%s seconds", handler_response)
- time.sleep(handler_response)
- return True
- def _send(self, request):
- return self.http_session.send(request)
- class EndpointCreator(object):
- def __init__(self, event_emitter):
- self._event_emitter = event_emitter
- def create_endpoint(self, service_model, region_name, endpoint_url,
- verify=None, response_parser_factory=None,
- timeout=DEFAULT_TIMEOUT,
- max_pool_connections=MAX_POOL_CONNECTIONS,
- http_session_cls=URLLib3Session,
- proxies=None,
- socket_options=None,
- client_cert=None,
- proxies_config=None):
- if not is_valid_endpoint_url(endpoint_url):
- raise ValueError("Invalid endpoint: %s" % endpoint_url)
- if proxies is None:
- proxies = self._get_proxies(endpoint_url)
- endpoint_prefix = service_model.endpoint_prefix
- logger.debug('Setting %s timeout as %s', endpoint_prefix, timeout)
- http_session = http_session_cls(
- timeout=timeout,
- proxies=proxies,
- verify=self._get_verify_value(verify),
- max_pool_connections=max_pool_connections,
- socket_options=socket_options,
- client_cert=client_cert,
- proxies_config=proxies_config
- )
- return Endpoint(
- endpoint_url,
- endpoint_prefix=endpoint_prefix,
- event_emitter=self._event_emitter,
- response_parser_factory=response_parser_factory,
- http_session=http_session
- )
- def _get_proxies(self, url):
- # We could also support getting proxies from a config file,
- # but for now proxy support is taken from the environment.
- return get_environ_proxies(url)
- def _get_verify_value(self, verify):
- # This is to account for:
- # https://github.com/kennethreitz/requests/issues/1436
- # where we need to honor REQUESTS_CA_BUNDLE because we're creating our
- # own request objects.
- # First, if verify is not None, then the user explicitly specified
- # a value so this automatically wins.
- if verify is not None:
- return verify
- # Otherwise use the value from REQUESTS_CA_BUNDLE, or default to
- # True if the env var does not exist.
- return os.environ.get('REQUESTS_CA_BUNDLE', True)
|