Repository: snowplow/snowplow-python-tracker

Issue:
Add customizable no-retry HTTP status codes
Add the ability to define whether to retry or not for a given HTTP status code

Discussion:
snowplowcla: Thanks for your pull request. Is this your first contribution to a Snowplow open source project? Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: Please visit https://docs.snowplowanalytics.com/docs/contributing/contributor-license-agreement/ to learn more and sign.

Once you've signed, please reply here (e.g. I signed it!) and we'll verify. Thanks.


Patch:
diff --git a/examples/snowplow_app.py b/examples/snowplow_app.py
index 7ce2c1d..9cc9f09 100644
--- a/examples/snowplow_app.py
+++ b/examples/snowplow_app.py
@@ -18,7 +18,8 @@ def main():
 
     collector_url = get_url_from_args()
     # Configure Emitter
-    emitter_config = EmitterConfiguration(batch_size=5)
+    custom_retry_codes = {500: False, 401: True}
+    emitter_config = EmitterConfiguration(batch_size=5, custom_retry_codes=custom_retry_codes)
 
     # Configure Tracker
     tracker_config = TrackerConfiguration(encode_base64=True)
diff --git a/snowplow_tracker/emitter_configuration.py b/snowplow_tracker/emitter_configuration.py
index c38823a..77f80bc 100644
--- a/snowplow_tracker/emitter_configuration.py
+++ b/snowplow_tracker/emitter_configuration.py
@@ -19,7 +19,7 @@
 #     License: Apache License Version 2.0
 # """
 
-from typing import Optional, Union, Tuple
+from typing import Optional, Union, Tuple, Dict
 from snowplow_tracker.typing import SuccessCallback, FailureCallback
 
 
@@ -31,7 +31,8 @@ class EmitterConfiguration(object):
         on_failure: Optional[FailureCallback] = None,
         byte_limit: Optional[int] = None,
         request_timeout: Optional[Union[float, Tuple[float, float]]] = None,
-        buffer_capacity: Optional[int] = None
+        buffer_capacity: Optional[int] = None,
+        custom_retry_codes: Dict[int, bool] = {}
     ) -> None:
         """
         Configuration for the emitter that sends events to the Snowplow collector.
@@ -52,6 +53,10 @@ class EmitterConfiguration(object):
                                  applies to both "connect" AND "read" timeout, or as tuple with two float values
                                  which specify the "connect" and "read" timeouts separately
         :type request_timeout:  float | tuple | None
+        :param  custom_retry_codes: Set custom retry rules for HTTP status codes received in emit responses from the Collector.
+                                    By default, retry will not occur for status codes 400, 401, 403, 410 or 422. This can be overridden here.
+                                    Note that 2xx codes will never retry as they are considered successful.
+        :type   custom_retry_codes: dict
         """
 
         self.batch_size = batch_size
@@ -60,6 +65,7 @@ class EmitterConfiguration(object):
         self.byte_limit = byte_limit
         self.request_timeout = request_timeout
         self.buffer_capacity = buffer_capacity
+        self.custom_retry_codes = custom_retry_codes
 
     @property
     def batch_size(self) -> Optional[int]:
@@ -144,4 +150,39 @@ class EmitterConfiguration(object):
             raise ValueError("buffer_capacity must greater than 0")
         if not isinstance(value, int) and value is not None:
             raise ValueError("buffer_capacity must be of type int")
-        self._buffer_capacity = value
\ No newline at end of file
+        self._buffer_capacity = value
+
+    @property
+    def custom_retry_codes(self) -> Dict[int, bool]:
+        """
+            Custom retry rules for HTTP status codes received in emit responses from the Collector.
+        """
+        return self._custom_retry_codes
+
+    @custom_retry_codes.setter
+    def custom_retry_codes(self, value: Dict[int, bool]):
+        self._custom_retry_codes = value
+    def set_retry_code(self, status_code: int, retry = True) -> bool:
+        """
+            Add a retry rule for HTTP status code received from emit responses from the Collector.
+            :param  status_code:    HTTP response code
+            :type   status_code:    int
+            :param  retry:  Set the status_code to retry (True) or not retry (False). Default is True
+            :type   retry:  bool
+        """
+        if not isinstance(status_code, int):
+            print("status_code must be of type int")
+            return False
+
+        if not isinstance(retry, bool):
+            print("retry must be of type bool")        
+            return False
+
+        if 200 <= status_code < 300:
+            print("custom_retry_codes should not include codes for succesful requests (2XX codes)")
+            return False
+
+        self.custom_retry_codes[status_code] = retry
+
+        return status_code in self.custom_retry_codes.keys()
+
diff --git a/snowplow_tracker/emitters.py b/snowplow_tracker/emitters.py
index 20acda8..3beaaec 100644
--- a/snowplow_tracker/emitters.py
+++ b/snowplow_tracker/emitters.py
@@ -25,7 +25,7 @@ import time
 import threading
 import requests
 import random
-from typing import Optional, Union, Tuple
+from typing import Optional, Union, Tuple, Dict
 from queue import Queue
 
 from snowplow_tracker.self_describing_json import SelfDescribingJson
@@ -71,6 +71,7 @@ class Emitter(object):
         request_timeout: Optional[Union[float, Tuple[float, float]]] = None,
         max_retry_delay_seconds: int = 60,
         buffer_capacity: int = 10000,
+        custom_retry_codes: Dict[int, bool] = {}
     ) -> None:
         """
         :param endpoint:    The collector URL. If protocol is not set in endpoint it will automatically set to "https://" - this is done automatically.
@@ -103,6 +104,10 @@ class Emitter(object):
         :param buffer_capacity: The maximum capacity of the event buffer. The default buffer capacity is 10 000 events.
                                 When the buffer is full new events are lost.
         :type buffer_capacity: int 
+        :param  custom_retry_codes: Set custom retry rules for HTTP status codes received in emit responses from the Collector.
+                                    By default, retry will not occur for status codes 400, 401, 403, 410 or 422. This can be overridden here.
+                                    Note that 2xx codes will never retry as they are considered successful.
+        :type   custom_retry_codes: dict
         """
         one_of(protocol, PROTOCOLS)
         one_of(method, METHODS)
@@ -137,6 +142,7 @@ class Emitter(object):
         self.retry_delay = 0
 
         self.buffer_capacity = buffer_capacity
+        self.custom_retry_codes = custom_retry_codes
         logger.info("Emitter initialized with endpoint " + self.endpoint)
 
     @staticmethod
@@ -373,6 +379,9 @@ class Emitter(object):
         if Emitter.is_good_status_code(status_code):
             return False
 
+        if status_code in self.custom_retry_codes.keys():
+            return self.custom_retry_codes[status_code]
+
         return status_code not in [400, 401, 403, 410, 422]
 
     def _set_retry_delay(self) -> None:
diff --git a/snowplow_tracker/snowplow.py b/snowplow_tracker/snowplow.py
index 8dd3164..946802e 100644
--- a/snowplow_tracker/snowplow.py
+++ b/snowplow_tracker/snowplow.py
@@ -82,6 +82,7 @@ class Snowplow:
             on_failure=emitter_config.on_failure,
             byte_limit=emitter_config.byte_limit,
             request_timeout=emitter_config.request_timeout,
+            custom_retry_codes=emitter_config.custom_retry_codes
         )
 
         tracker = Tracker(


Test patch:
diff --git a/snowplow_tracker/test/unit/test_emitters.py b/snowplow_tracker/test/unit/test_emitters.py
index c879091..439b66a 100644
--- a/snowplow_tracker/test/unit/test_emitters.py
+++ b/snowplow_tracker/test/unit/test_emitters.py
@@ -475,3 +475,36 @@ class TestEmitters(unittest.TestCase):
         
         mok_failure.assert_called_once_with(0, evBuffer)
         mok_success.assert_not_called()
+
+    @mock.patch('snowplow_tracker.Emitter.http_post')
+    def test_send_events_post_custom_retry(self, mok_http_post: Any) -> None:
+        mok_http_post.side_effect = mocked_http_response_failure
+        mok_success = mock.Mock(return_value="success mocked")
+        mok_failure = mock.Mock(return_value="failure mocked")
+
+        e = Emitter('0.0.0.0', batch_size=10, on_success=mok_success, on_failure=mok_failure, custom_retry_codes={400: True})
+        evBuffer = [{"a": "aa"}, {"b": "bb"}, {"c": "cc"}]
+        e.send_events(evBuffer)
+        
+        mok_http_post.side_effect = mocked_http_response_success
+        time.sleep(5)
+
+        mok_failure.assert_called_with(0, evBuffer)
+        mok_success.assert_called_with(evBuffer)
+
+    @mock.patch('snowplow_tracker.Emitter.http_get')
+    def test_send_events_get_custom_retry(self, mok_http_get: Any) -> None:
+        mok_http_get.side_effect = mocked_http_response_failure
+        mok_success = mock.Mock(return_value="success mocked")
+        mok_failure = mock.Mock(return_value="failure mocked")
+
+        e = Emitter('0.0.0.0', method='get',batch_size=10, on_success=mok_success, on_failure=mok_failure, custom_retry_codes={400: True})
+        evBuffer = [{"a": "aa"}, {"b": "bb"}, {"c": "cc"}]
+        e.send_events(evBuffer)
+        
+        mok_http_get.side_effect = mocked_http_response_success
+        time.sleep(5)
+
+        mok_failure.assert_called_with(0, evBuffer)
+        mok_success.assert_called_with(evBuffer)
+