Tdl 681 add retry mechanism

tap_klaviyo/__init__.py

@@ -141,7 +141,7 @@ def discover(api_key):
 
 
 def do_discover(api_key):
-    print(json.dumps(discover(api_key), sys.stdout, indent=2))
+    print(json.dumps(discover(api_key), indent=2))
 
 def main():
 

tap_klaviyo/utils.py

@@ -3,13 +3,67 @@
 import singer
 from singer import metrics
 import requests
+import backoff
+import json
+import simplejson
 
 DATETIME_FMT = "%Y-%m-%dT%H:%M:%SZ"
 
-
 session = requests.Session()
 logger = singer.get_logger()
 
+class KlaviyoError(Exception):
+    pass
+
+class KlaviyoNotFoundError(KlaviyoError):
+    pass
+
+class KlaviyoBadRequestError(KlaviyoError):
+    pass
+
+class KlaviyoUnauthorizedError(KlaviyoError):
+    pass
+
+class KlaviyoForbiddenError(KlaviyoError):
+    pass
+
+class KlaviyoInternalServiceError(KlaviyoError):
+    pass
+
+ERROR_CODE_EXCEPTION_MAPPING = {
+    400: {
+        "raise_exception": KlaviyoBadRequestError,
+        "message": "Request is missing or has a bad parameter."
+    },
+    401: {
+        "raise_exception": KlaviyoUnauthorizedError,
+        "message": "Invalid authorization credentials."
+    },
+    403: {
+        "raise_exception": KlaviyoForbiddenError,
+        "message": "Request is missing or has an invalid API key."
+    },
+    404: {
+        "raise_exception": KlaviyoNotFoundError,
+        "message": "The requested resource doesn't exist."
+    },
+    500: {
+        "raise_exception": KlaviyoInternalServiceError,
+        "message": "Something is wrong on Klaviyo's end."
+    }
+}
+
+def raise_for_error(response):   
+    try:
+        json_resp = response.json()
+    except Exception:
+        json_resp = {}
+
+    error_code = response.status_code
+    message_text = json_resp.get("message", ERROR_CODE_EXCEPTION_MAPPING.get(error_code, {}).get("message", "Unknown Error"))
+    message = "HTTP-error-code: {}, Error: {}".format(error_code, message_text)
+    exc = ERROR_CODE_EXCEPTION_MAPPING.get(error_code, {}).get("raise_exception", KlaviyoError)
+    raise exc(message)
 
 def dt_to_ts(dt):
     return int(time.mktime(datetime.datetime.strptime(
@@ -50,12 +104,16 @@ def get_starting_point(stream, state, start_date):
 def get_latest_event_time(events):
     return ts_to_dt(int(events[-1]['timestamp'])) if len(events) else None
 
-
+@backoff.on_exception(backoff.expo, KlaviyoError, max_tries=3)
 def authed_get(source, url, params):
     with metrics.http_request_timer(source) as timer:
         resp = session.request(method='get', url=url, params=params)
-        timer.tags[metrics.Tag.http_status_code] = resp.status_code
-        return resp
+
+        if resp.status_code != 200:
+            raise_for_error(resp)
+        else:
+            timer.tags[metrics.Tag.http_status_code] = resp.status_code
+            return resp
 
 
 def get_all_using_next(stream, url, api_key, since=None):
@@ -80,7 +138,7 @@ def get_all_pages(source, url, api_key):
         else:
             break
 
-
+@backoff.on_exception(backoff.expo, simplejson.scanner.JSONDecodeError, max_tries=3)
 def get_incremental_pull(stream, endpoint, state, api_key, start_date):
     latest_event_time = get_starting_point(stream, state, start_date)
 
@@ -104,12 +162,14 @@ def get_incremental_pull(stream, endpoint, state, api_key, start_date):
 
     return state
 
-
+@backoff.on_exception(backoff.expo, simplejson.scanner.JSONDecodeError, max_tries=3)
 def get_full_pulls(resource, endpoint, api_key):
+
     with metrics.record_counter(resource['stream']) as counter:
         for response in get_all_pages(resource['stream'], endpoint, api_key):
+
             records = response.json().get('data')
 
             counter.increment(len(records))
 
-            singer.write_records(resource['stream'], records)
+            singer.write_records(resource['stream'], records)

tests/test_backoff.py

@@ -0,0 +1,38 @@
+import tap_klaviyo.utils as utils_
+import unittest
+from unittest import mock
+import simplejson
+import singer
+from singer import metrics
+import json
+import requests
+
+class TestBackoff(unittest.TestCase):
+
+    @mock.patch('requests.Session.request')
+    def test_httperror(self, mocked_session):
+
+        mock_resp = mock.Mock()
+        klaviyo_error = utils_.KlaviyoError()
+        mock_resp.raise_for_error.side_effect = klaviyo_error
+        mocked_session.return_value = mock_resp
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError:
+            pass
+
+        self.assertEquals(mocked_session.call_count, 3)
+
+    @mock.patch("tap_klaviyo.utils.get_all_pages")
+    def test_jsondecode(self, mock1):
+
+        mock1.return_value = ["abc"]
+
+        data = {'stream': 'lists'}
+        try:
+            utils_.get_full_pulls(data, "", "")
+        except simplejson.scanner.JSONDecodeError:
+            pass
+
+        self.assertEquals(mock1.call_count, 3)

tests/test_exception_handling.py

@@ -0,0 +1,116 @@
+import tap_klaviyo.utils as utils_
+import unittest
+from unittest import mock
+import requests
+import singer
+
+class Mockresponse:
+    def __init__(self, resp, status_code, content=None, headers=None, raise_error=False):
+        self.json_data = resp
+        self.status_code = status_code
+        self.content = content
+        self.headers = headers
+        self.raise_error = raise_error
+
+    def json(self):
+            return self.json_data
+
+def klaviyo_400_error(*args, **kwargs):
+    json_str = {}
+
+    return Mockresponse(json_str, 400, raise_error=True)
+
+def klaviyo_401_error(*args, **kwargs):
+    json_str = {}
+
+    return Mockresponse(json_str, 401, raise_error=True)
+
+def klaviyo_403_error(*args, **kwargs):
+    json_str = {}
+
+    return Mockresponse(json_str, 403, raise_error=True)
+
+def klaviyo_403_error_wrong_api_key(*args, **kwargs):
+    json_str = {
+    "status": 403,
+    "message": "The API key specified is invalid."}
+
+    return Mockresponse(json_str, 403, raise_error=True)
+
+def klaviyo_403_error_missing_api_key(*args, **kwargs):
+    json_str = {
+    "status": 403,
+    "message": "You must specify an API key to make requests."}
+
+    return Mockresponse(json_str, 403, raise_error=True)
+
+def klaviyo_404_error(*args, **kwargs):
+    json_str = {}
+
+    return Mockresponse(json_str, 404, raise_error=True)
+
+def klaviyo_500_error(*args, **kwargs):
+    json_str = {}
+
+    return Mockresponse(json_str, 500, raise_error=True)
+
+
+class TestBackoff(unittest.TestCase):
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_400_error)
+    def test_400_error(self, klaviyo_400_error):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 400, Error: Request is missing or has a bad parameter.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_401_error)
+    def test_401_error(self, klaviyo_401_error):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 401, Error: Invalid authorization credentials.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_403_error)
+    def test_403_error(self, klaviyo_403_error):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            with open("abc.txt", "w") as f:
+                f.write(str(e))
+            self.assertEquals(str(e), "HTTP-error-code: 403, Error: Request is missing or has an invalid API key.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_403_error_wrong_api_key)
+    def test_403_error_wrong_api_key(self, klaviyo_403_error_wrong_api_key):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 403, Error: The API key specified is invalid.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_403_error_missing_api_key)
+    def test_403_error_missing_api_key(self, klaviyo_403_error_missing_api_key):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 403, Error: You must specify an API key to make requests.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_404_error)
+    def test_404_error(self, klaviyo_404_error):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 404, Error: The requested resource doesn't exist.")
+
+    @mock.patch('requests.Session.request', side_effect=klaviyo_500_error)
+    def test_500_error(self, klaviyo_500_error):
+
+        try:
+            utils_.authed_get("", "", "")
+        except utils_.KlaviyoError as e:
+            self.assertEquals(str(e), "HTTP-error-code: 500, Error: Something is wrong on Klaviyo's end.")