Stop using requests in favour of botocore & urllib3 (#580)

Author: garrettheel

.travis.yml

@@ -4,7 +4,6 @@ python:
   - "3.6"
   - "3.5"
   - "2.7"
-  - "2.6"
   - "pypy"
 
 

README.rst

@@ -228,7 +228,7 @@ Want to backup and restore a table? No problem.
 Features
 ========
 
-* Python 3.3, 3.4, 2.6, and 2.7 support
+* Python >= 3.3, and 2.7 support
 * An ORM-like interface with query and scan filters
 * Compatible with DynamoDB Local
 * Supports the entire DynamoDB API

docs/release_notes.rst

@@ -1,6 +1,29 @@
 Release Notes
 =============
 
+v4.0.0a1
+--------
+
+:date: 2019-04-10
+
+NB: This is an alpha release and these notes are subject to change.
+
+This is major release and contains breaking changes. Please read the notes below carefully.
+
+Given that ``botocore`` has moved to using ``urllib3`` directly for making HTTP requests, we'll be doing the same (via ``botocore``). This means the following:
+
+* The ``session_cls`` option is no longer supported. S
+* The ``request_timeout_seconds`` parameter is no longer supported. ``connect_timeout_seconds`` and ``read_timeout_seconds`` are now available instead.
+    * Note that the timeout for connection and read are now ``15`` and ``30`` seconds respectively. This represents a change from the previous ``60`` second combined ``requests`` timeout.
+* *Wrapped* exceptions (i.e ``exc.cause``) that were from ``requests.exceptions`` will now be comparable ones from ``botocore.exceptions`` instead.
+
+Other changes in this release:
+
+* Python 2.6 is no longer supported. 4.x.x will likely be the last major release to support Python 2.7, given the upcoming EOL.
+* Added the ``max_pool_connection`` and ``extra_headers`` settings to replace common use cases for ``session_cls``
+* Added support for `moto <https://github.com/spulec/moto>`_ through implementing the botocore "before-send" hook. Other botocore hooks remain unimplemented.
+
+
 v3.3.3
 ------
 
@@ -10,12 +33,12 @@ This is a backwards compatible, minor release.
 
 Fixes in this release:
 
- * Legacy boolean attribute migration fix. (#538)
- * Correctly package type stubs. (#585)
+* Legacy boolean attribute migration fix. (#538)
+* Correctly package type stubs. (#585)
 
 Contributors to this release:
 
- * @vo-va
+* @vo-va
 
 
 v3.3.2
@@ -27,7 +50,7 @@ This is a backwards compatible, minor release.
 
 Changes in this release:
 
- * Built-in support for mypy type stubs, superseding those in python/typeshed. (#537)
+* Built-in support for mypy type stubs, superseding those in python/typeshed. (#537)
 
 
 v3.3.1
@@ -89,7 +112,7 @@ Contributors to this release:
 * @nicysneiros
 * @jcomo
 * @kevgliss
-* @asottile 
+* @asottile
 * @harleyk
 * @betamoo
 

docs/settings.rst

@@ -9,13 +9,20 @@ Settings reference
 
 Here is a complete list of settings which control default PynamoDB behavior.
 
+connect_timeout_seconds
+-----------------------
+
+Default: ``15``
+
+The time in seconds till a ``ConnectTimeoutError`` is thrown when attempting to make a connection.
+
 
-request_timeout_seconds
+read_timeout_seconds
 -----------------------
 
-Default: ``60``
+Default: ``30``
 
-The default timeout for HTTP requests in seconds.
+The time in seconds till a ``ReadTimeoutError`` is thrown when attempting to read from a connection.
 
 
 max_retry_attempts
@@ -44,15 +51,24 @@ Default: ``"us-east-1"``
 The default AWS region to connect to.
 
 
-session_cls
------------
+max_pool_connections
+--------------------
+
+Default: ``10``
+
+The maximum number of connections to keep in a connection pool.
+
 
-Default: ``botocore.vendored.requests.Session``
+extra_headers
+--------------------
 
-A class which implements the Session_ interface from requests, used for making API requests
-to DynamoDB.
+Default: ``None``
+
+A dictionary of headers that should be added to every request. This is only useful
+when interfacing with DynamoDB through a proxy, where headers are stripped by the
+proxy before forwarding along. Failure to strip these headers before sending to AWS
+will result in an ``InvalidSignatureException`` due to request signing.
 
-.. _Session: http://docs.python-requests.org/en/master/api/#request-sessions
 
 allow_rate_limited_scan_without_consumed_capacity
 -------------------------------------------------
@@ -71,16 +87,3 @@ Overriding settings
 Default settings may be overridden by providing a Python module which exports the desired new values.
 Set the ``PYNAMODB_CONFIG`` environment variable to an absolute path to this module or write it to
 ``/etc/pynamodb/global_default_settings.py`` to have it automatically discovered.
-
-See an example of specifying a custom ``session_cls`` to configure the connection pool below.
-
-.. code-block:: python
-
-    from botocore.vendored import requests
-    from botocore.vendored.requests import adapters
-
-    class CustomPynamoSession(requests.Session):
-        super(CustomPynamoSession, self).__init__()
-        self.mount('http://', adapters.HTTPAdapter(pool_maxsize=100))
-
-    session_cls = CustomPynamoSession

pynamodb/__init__.py

@@ -7,4 +7,4 @@
 """
 __author__ = 'Jharrod LaFon'
 __license__ = 'MIT'
-__version__ = '3.3.3'
+__version__ = '4.0.0a1'

pynamodb/connection/base.py

@@ -3,17 +3,22 @@
 """
 from __future__ import division
 
+import json
 import logging
 import math
 import random
+import sys
 import time
 import uuid
 import warnings
 from base64 import b64decode
 from threading import local
 
 import six
+import botocore.client
+import botocore.exceptions
 from botocore.client import ClientError
+from botocore.hooks import first_non_none_response
 from botocore.exceptions import BotoCoreError
 from botocore.session import get_session
 from botocore.vendored import requests
@@ -222,27 +227,28 @@ class Connection(object):
     A higher level abstraction over botocore
     """
 
-    def __init__(self, region=None, host=None, session_cls=None,
-                 request_timeout_seconds=None, max_retry_attempts=None, base_backoff_ms=None):
+    def __init__(self, region=None, host=None,
+                 read_timeout_seconds=None, connect_timeout_seconds=None,
+                 max_retry_attempts=None, base_backoff_ms=None,
+                 max_pool_connections=None, extra_headers=None):
         self._tables = {}
         self.host = host
         self._local = local()
-        self._requests_session = None
         self._client = None
         if region:
             self.region = region
         else:
             self.region = get_settings_value('region')
 
-        if session_cls:
-            self.session_cls = session_cls
+        if connect_timeout_seconds is not None:
+            self._connect_timeout_seconds = connect_timeout_seconds
         else:
-            self.session_cls = get_settings_value('session_cls')
+            self._connect_timeout_seconds = get_settings_value('connect_timeout_seconds')
 
-        if request_timeout_seconds is not None:
-            self._request_timeout_seconds = request_timeout_seconds
+        if read_timeout_seconds is not None:
+            self._read_timeout_seconds = read_timeout_seconds
         else:
-            self._request_timeout_seconds = get_settings_value('request_timeout_seconds')
+            self._read_timeout_seconds = get_settings_value('read_timeout_seconds')
 
         if max_retry_attempts is not None:
             self._max_retry_attempts_exception = max_retry_attempts
@@ -254,6 +260,16 @@ def __init__(self, region=None, host=None, session_cls=None,
         else:
             self._base_backoff_ms = get_settings_value('base_backoff_ms')
 
+        if max_pool_connections is not None:
+            self._max_pool_connections = max_pool_connections
+        else:
+            self._max_pool_connections = get_settings_value('max_pool_connections')
+
+        if extra_headers is not None:
+            self._extra_headers = extra_headers
+        else:
+            self._extra_headers = get_settings_value('extra_headers')
+
     def __repr__(self):
         return six.u("Connection<{0}>".format(self.client.meta.endpoint_url))
 
@@ -276,24 +292,11 @@ def _log_error(self, operation, response):
         log.error("%s failed with status: %s, message: %s",
                   operation, response.status_code,response.content)
 
-    def _create_prepared_request(self, request_dict, operation_model):
-        """
-        Create a prepared request object from request_dict, and operation_model
-        """
-        boto_prepared_request = self.client._endpoint.create_request(request_dict, operation_model)
-
-        # The call requests_session.send(final_prepared_request) ignores the headers which are
-        # part of the request session. In order to include the requests session headers inside
-        # the request, we create a new request object, and call prepare_request with the newly
-        # created request object
-        raw_request_with_params = Request(
-            boto_prepared_request.method,
-            boto_prepared_request.url,
-            data=boto_prepared_request.body,
-            headers=boto_prepared_request.headers
-        )
-
-        return self.requests_session.prepare_request(raw_request_with_params)
+    def _create_prepared_request(self, params, operation_model):
+        prepared_request = self.client._endpoint.create_request(params, operation_model)
+        if self._extra_headers is not None:
+            prepared_request.headers.update(self._extra_headers)
+        return prepared_request
 
     def dispatch(self, operation_name, operation_kwargs):
         """
@@ -341,32 +344,47 @@ def _make_api_call(self, operation_name, operation_kwargs):
         operation_model = self.client._service_model.operation_model(operation_name)
         request_dict = self.client._convert_to_request_dict(
             operation_kwargs,
-            operation_model
+            operation_model,
         )
-        prepared_request = self._create_prepared_request(request_dict, operation_model)
 
         for i in range(0, self._max_retry_attempts_exception + 1):
             attempt_number = i + 1
             is_last_attempt_for_exceptions = i == self._max_retry_attempts_exception
 
-            response = None
+            http_response = None
+            prepared_request = None
             try:
-                proxies = getattr(self.client._endpoint, "proxies", None)
-                # After the version 1.11.0 of botocore this field is no longer available here
-                if proxies is None:
-                    proxies = self.client._endpoint.http_session._proxy_config._proxies
-
-                response = self.requests_session.send(
-                    prepared_request,
-                    timeout=self._request_timeout_seconds,
-                    proxies=proxies,
-                )
-                data = response.json()
-            except (requests.RequestException, ValueError) as e:
+                if prepared_request is not None:
+                    # 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.
+                    prepared_request.reset_stream()
+
+                # Create a new request for each retry (including a new signature).
+                prepared_request = self._create_prepared_request(request_dict, operation_model)
+
+                # Implement the before-send event from botocore
+                event_name = 'before-send.dynamodb.{}'.format(operation_model.name)
+                event_responses = self.client._endpoint._event_emitter.emit(event_name, request=prepared_request)
+                event_response = first_non_none_response(event_responses)
+
+                if event_response is None:
+                    http_response = self.client._endpoint.http_session.send(prepared_request)
+                else:
+                    http_response = event_response
+                    is_last_attempt_for_exceptions = True  # don't retry if we have an event response
+
+                # json.loads accepts bytes in >= 3.6.0
+                if sys.version_info < (3, 6, 0):
+                    data = json.loads(http_response.text)
+                else:
+                    data = json.loads(http_response.content)
+            except (ValueError, botocore.exceptions.HTTPClientError, botocore.exceptions.ConnectionError) as e:
                 if is_last_attempt_for_exceptions:
                     log.debug('Reached the maximum number of retry attempts: %s', attempt_number)
-                    if response:
-                        e.args += (str(response.content),)
+                    if http_response:
+                        e.args += (http_response.text,)
                     raise
                 else:
                     # No backoff for fast-fail exceptions that likely failed at the frontend
@@ -379,14 +397,16 @@ def _make_api_call(self, operation_name, operation_kwargs):
                     )
                     continue
 
-            if response.status_code >= 300:
+            status_code = http_response.status_code
+            headers = http_response.headers
+            if status_code >= 300:
                 # Extract error code from __type
                 code = data.get('__type', '')
                 if '#' in code:
                     code = code.rsplit('#', 1)[1]
                 botocore_expected_format = {'Error': {'Message': data.get('message', ''), 'Code': code}}
                 verbose_properties = {
-                    'request_id': response.headers.get('x-amzn-RequestId')
+                    'request_id': headers.get('x-amzn-RequestId')
                 }
 
                 if 'RequestItems' in operation_kwargs:
@@ -401,7 +421,7 @@ def _make_api_call(self, operation_name, operation_kwargs):
                     if is_last_attempt_for_exceptions:
                         log.debug('Reached the maximum number of retry attempts: %s', attempt_number)
                         raise
-                    elif response.status_code < 500 and code != 'ProvisionedThroughputExceededException':
+                    elif status_code < 500 and code != 'ProvisionedThroughputExceededException':
                         # We don't retry on a ConditionalCheckFailedException or other 4xx (except for
                         # throughput related errors) because we assume they will fail in perpetuity.
                         # Retrying when there is already contention could cause other problems
@@ -470,15 +490,6 @@ def session(self):
             self._local.session = get_session()
         return self._local.session
 
-    @property
-    def requests_session(self):
-        """
-        Return a requests session to execute prepared requests using the same pool
-        """
-        if self._requests_session is None:
-            self._requests_session = self.session_cls()
-        return self._requests_session
-
     @property
     def client(self):
         """
@@ -489,7 +500,11 @@ def client(self):
         # if the client does not have credentials, we create a new client
         # otherwise the client is permanently poisoned in the case of metadata service flakiness when using IAM roles
         if not self._client or (self._client._request_signer and not self._client._request_signer._credentials):
-            self._client = self.session.create_client(SERVICE_NAME, self.region, endpoint_url=self.host)
+            config = botocore.client.Config(
+                connect_timeout=self._connect_timeout_seconds,
+                read_timeout=self._read_timeout_seconds,
+                max_pool_connections=self._max_pool_connections)
+            self._client = self.session.create_client(SERVICE_NAME, self.region, endpoint_url=self.host, config=config)
         return self._client
 
     def get_meta_table(self, table_name, refresh=False):