Merge pull request #88 from customerio/option_to_control_connection_pooling

Adds an additional named parameter to the CustomerIO and APIClient in…

Author: scraswell

README.md

@@ -226,6 +226,17 @@ response = client.send_email(request)
 print(response)
 ```
 
+## Notes
+- The Customer.io Python SDK depends on the [`Requests`](https://pypi.org/project/requests/) library which includes [`urllib3`](https://pypi.org/project/urllib3/) as a transitive dependency.  The [`Requests`](https://pypi.org/project/requests/) library leverages connection pooling defined in [`urllib3`](https://pypi.org/project/urllib3/).  [`urllib3`](https://pypi.org/project/urllib3/) only attempts to retry invocations of `HTTP` methods which are understood to be idempotent (See: [`Retry.DEFAULT_ALLOWED_METHODS`](https://github.com/urllib3/urllib3/blob/main/src/urllib3/util/retry.py#L184)).  Since the `POST` method is not considered to be idempotent, any invocations which require `POST` are not retried.
+
+- It is possible to have the Customer.io Python SDK effectively *disable* connection pooling by passing a named initialization parameter `use_connection_pooling` to either the `APIClient` class or `CustomerIO` class.  Setting this parameter to `False` (default: `True`) causes the [`Session`](https://github.com/psf/requests/blob/main/requests/sessions.py#L355) to be initialized and discarded after each request.  If you are experiencing integration issues where the cause is reported as `Connection Reset by Peer`, this may correct the problem.  It will, however, impose a slight performance penalty as the TCP connection set-up and tear-down will now occur for each request.
+
+### Usage Example Disabling Connection Pooling
+```python
+from customerio import CustomerIO, Regions
+cio = CustomerIO(site_id, api_key, region=Regions.US, use_connection_pooling=False)
+```
+
 ## Running tests
 
 Changes to the library can be tested by running `make test` from the parent directory.

customerio/__version__.py

@@ -1,4 +1,4 @@
-VERSION = (1, 5, 0, 'final', 0)
+VERSION = (1, 6, 0, 'final', 0)
 
 def get_version():
     version = '%s.%s' % (VERSION[0], VERSION[1])

customerio/api.py

@@ -7,23 +7,27 @@
 from .regions import Regions, Region
 
 class APIClient(ClientBase):
-    def __init__(self, key, url=None, region=Regions.US, retries=3, timeout=10, backoff_factor=0.02):
+    def __init__(self, key, url=None, region=Regions.US, retries=3, timeout=10, backoff_factor=0.02, use_connection_pooling=True):
         if not isinstance(region, Region):
             raise CustomerIOException('invalid region provided')
 
+        self.key = key
         self.url = url or 'https://{host}'.format(host=region.api_host)
         ClientBase.__init__(self, retries=retries,
-                            timeout=timeout, backoff_factor=backoff_factor)
-
-        self.http.headers['Authorization'] = "Bearer {key}".format(key=key)
+                            timeout=timeout, backoff_factor=backoff_factor, use_connection_pooling=use_connection_pooling)
 
     def send_email(self, request):
         if isinstance(request, SendEmailRequest):
             request = request._to_dict()
         resp = self.send_request('POST', self.url + "/v1/send/email", request)
         return json.loads(resp)
 
+    # builds the session.
+    def _build_session(self):
+        session = super()._build_session()
+        session.headers['Authorization'] = "Bearer {key}".format(key=self.key)
 
+        return session
 
 class SendEmailRequest(object):
     '''An object with all the options avaiable for triggering a transactional message'''

customerio/client_base.py

@@ -3,6 +3,7 @@
 """
 from __future__ import division
 from datetime import datetime, timezone
+import logging
 import math
 
 from requests import Session
@@ -15,24 +16,32 @@ class CustomerIOException(Exception):
     pass
 
 class ClientBase(object):
-    def __init__(self, retries=3, timeout=10, backoff_factor=0.02):
+    def __init__(self, retries=3, timeout=10, backoff_factor=0.02, use_connection_pooling=True):
         self.timeout = timeout
         self.retries = retries
+        self.backoff_factor = backoff_factor
+        self.use_connection_pooling = use_connection_pooling
+        self._current_session = None
 
-        self.http = Session()
-        self.http.headers['User-Agent'] = "Customer.io Python Client/{version}".format(version=ClientVersion)
+    @property
+    def http(self):
+        if self._current_session is None:
+            self._current_session = self._get_session()
 
-        # Retry request a number of times before raising an exception
-        # also define backoff_factor to delay each retry
-        self.http.mount('https://', HTTPAdapter(max_retries=Retry(
-            total=retries, backoff_factor=backoff_factor)))
+        return self._current_session
 
     def send_request(self, method, url, data):
         '''Dispatches the request and returns a response'''
 
         try:
             response = self.http.request(
                 method, url=url, json=self._sanitize(data), timeout=self.timeout)
+
+            result_status = response.status_code
+            if result_status != 200:
+                raise CustomerIOException('%s: %s %s %s' % (result_status, url, data, response.text))
+            return response.text
+
         except Exception as e:
             # Raise exception alerting user that the system might be
             # experiencing an outage and refer them to system status page.
@@ -42,10 +51,8 @@ def send_request(self, method, url, data):
             '''.format(klass=type(e), message=e, count=self.retries)
             raise CustomerIOException(message)
 
-        result_status = response.status_code
-        if result_status != 200:
-            raise CustomerIOException('%s: %s %s %s' % (result_status, url, data, response.text))
-        return response.text
+        finally:
+            self._close()
 
     def _sanitize(self, data):
         for k, v in data.items():
@@ -69,3 +76,38 @@ def _stringify_list(self, customer_ids):
                 raise CustomerIOException(
                     'customer_ids cannot be {type}'.format(type=type(v)))
         return customer_string_ids
+
+    # gets a session based on whether we want pooling or not.  If no pooling is desired, we create a new session each time.
+    def _get_session(self):
+        if (self.use_connection_pooling):
+            if (self._current_session is None):
+                self._current_session = self._build_session()
+
+            # if we're using pooling, return the existing session.
+            logging.debug("Using existing session...")
+            return self._current_session
+        else:
+            # if we're not using pooling, build a new session.
+            logging.debug("Creating new session...")
+            self._current_session = self._build_session()
+        return self._current_session
+
+    # builds the session.
+    def _build_session(self):
+        session = Session()
+        session.headers['User-Agent'] = "Customer.io Python Client/{version}".format(version=ClientVersion)
+
+        # Retry request a number of times before raising an exception
+        # also define backoff_factor to delay each retry
+        session.mount(
+            'https://',
+            HTTPAdapter(max_retries=Retry(total=self.retries, backoff_factor=self.backoff_factor)))
+
+        return session
+
+    # closes the session if we're not using connection pooling.
+    def _close(self):
+        # if we're not using pooling; clean up the resources.
+        if (not self.use_connection_pooling):
+            self._current_session.close()
+            self._current_session = None

customerio/track.py

@@ -10,22 +10,27 @@
 from customerio.constants import CIOID, EMAIL, ID
 
 class CustomerIO(ClientBase):
-    def __init__(self, site_id=None, api_key=None, host=None, region=Regions.US, port=None, url_prefix=None, json_encoder=None, retries=3, timeout=10, backoff_factor=0.02):
+    def __init__(self, site_id=None, api_key=None, host=None, region=Regions.US, port=None, url_prefix=None, json_encoder=None, retries=3, timeout=10, backoff_factor=0.02, use_connection_pooling=True):
         if not isinstance(region, Region):
             raise CustomerIOException('invalid region provided')
 
         self.host = host or region.track_host
         self.port = port or 443
         self.url_prefix = url_prefix or '/api/v1'
+        self.api_key = api_key
+        self.site_id = site_id
 
         if json_encoder is not None:
             warnings.warn(
                 "With the switch to using requests library the `json_encoder` param is no longer used.", DeprecationWarning)
 
         self.setup_base_url()
-        ClientBase.__init__(self, retries=retries,
-                            timeout=timeout, backoff_factor=backoff_factor)
-        self.http.auth = (site_id, api_key)
+        ClientBase.__init__(
+            self,
+            retries=retries,
+            timeout=timeout,
+            backoff_factor=backoff_factor,
+            use_connection_pooling=use_connection_pooling)
 
     def _url_encode(self, id):
         return quote(str(id), safe='')
@@ -209,3 +214,10 @@ def merge_customers(self, primary_id_type, primary_id, secondary_id_type, second
             }
         }
         self.send_request('POST', url, post_data)
+
+    # builds the session.
+    def _build_session(self):
+        session = super()._build_session()
+        session.auth = (self.site_id, self.api_key)
+
+        return session