Merge branch 'master' into add-talk-api

Author: leo-schick

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 1.7.5
+  * Added support for backoff and retry for error 409 [#107](https://github.com/singer-io/tap-zendesk/pull/107)
+  * Code Formatting [#107](https://github.com/singer-io/tap-zendesk/pull/107)
 ## 1.7.4
   * Request Timeout Implementation [#79](https://github.com/singer-io/tap-zendesk/pull/79)
 ## 1.7.3

setup.py

@@ -3,7 +3,7 @@
 from setuptools import setup
 
 setup(name='tap-zendesk',
-      version='1.7.4',
+      version='1.7.5',
       description='Singer.io tap for extracting data from the Zendesk API',
       author='Stitch',
       url='https://singer.io',

tap_zendesk/http.py

@@ -109,6 +109,16 @@ def is_fatal(exception):
 
     return 400 <=status_code < 500
 
+def should_retry_error(exception):
+    """
+        Return true if exception is required to retry otherwise return false
+    """
+    if isinstance(exception, ZendeskConflictError):
+        return True
+    if isinstance(exception,Exception) and isinstance(exception.args[0][1],ConnectionResetError):
+        return True
+    return False
+
 def raise_for_error(response):
     """ Error handling method which throws custom error. Class for each error defined above which extends `ZendeskError`.
     This method map the status code with `ERROR_CODE_EXCEPTION_MAPPING` dictionary and accordingly raise the error.
@@ -130,6 +140,10 @@ def raise_for_error(response):
             response.status_code, {}).get("raise_exception", ZendeskError)
         raise exc(message, response) from None
 
+@backoff.on_exception(backoff.expo,
+                      (ZendeskConflictError),
+                      max_tries=10,
+                      giveup=lambda e: not should_retry_error(e))
 @backoff.on_exception(backoff.expo,
                       (HTTPError, ZendeskError), # Added support of backoff for all unhandled status codes.
                       max_tries=10,
@@ -158,7 +172,6 @@ def get_cursor_based(url, access_token, request_timeout, cursor=None, **kwargs):
 
     if cursor:
         params['page[after]'] = cursor
-
     response = call_api(url, request_timeout, params=params, headers=headers)
     response_json = response.json()
 

test/test_automatic_fields.py

@@ -2,12 +2,15 @@
 import tap_tester.runner as runner
 from base import ZendeskTest
 
+# BUG https://jira.talendforge.org/browse/TDL-19428
+#     [tap-zendesk] Consistently replicating duplicate `organizations` record
+
 class ZendeskAutomaticFields(ZendeskTest):
     """
-    Ensure running the tap with all streams selected and all fields deselected results in the replication of just the 
+    Ensure running the tap with all streams selected and all fields deselected results in the replication of just the
     primary keys and replication keys (automatic fields).
     """
-    
+
     def name(self):
         return "zendesk_automatic_fields"
 
@@ -17,7 +20,7 @@ def test_run(self):
         Verify that only the automatic fields are sent to the target.
         Verify that all replicated records have unique primary key values.
         """
-        
+
         streams_to_test = self.expected_check_streams()
 
         conn_id = connections.ensure_connection(self)
@@ -34,14 +37,14 @@ def test_run(self):
 
         record_count_by_stream = self.run_and_verify_sync(conn_id)
         synced_records = runner.get_records_from_target_output()
-        
+
         for stream in streams_to_test:
             with self.subTest(stream=stream):
 
                 # expected values
                 expected_keys = self.expected_automatic_fields().get(stream)
                 expected_primary_keys = self.expected_primary_keys()[stream]
-                
+
                 # collect actual values
                 data = synced_records.get(stream, {})
                 record_messages_keys = [set(row['data'].keys())
@@ -50,7 +53,7 @@ def test_run(self):
                                        for message in data.get('messages', [])
                                        if message.get('action') == 'upsert']
                 unique_primary_keys_list = set(primary_keys_list)
-                
+
                 # Verify that you get some records for each stream
                 self.assertGreater(
                     record_count_by_stream.get(stream, -1), 0,
@@ -59,8 +62,10 @@ def test_run(self):
                 # Verify that only the automatic fields are sent to the target
                 for actual_keys in record_messages_keys:
                     self.assertSetEqual(expected_keys, actual_keys)
-                    
-                #Verify that all replicated records have unique primary key values.
-                self.assertEqual(len(primary_keys_list), 
-                                    len(unique_primary_keys_list), 
-                                    msg="Replicated record does not have unique primary key values.")
+
+                # Verify that all replicated records have unique primary key values.
+                if stream == 'organizations': # BUG_TDL-19428
+                    continue # skipping
+                self.assertEqual(len(primary_keys_list),
+                                 len(unique_primary_keys_list),
+                                 msg="Replicated record does not have unique primary key values.")