TDL-18879: Clear offset when max_skip error is encountered and return 0 if skip is greater than 250k in the current state.

tap_closeio/context.py

@@ -46,7 +46,12 @@ def set_bookmark(self, path, val):
 
     def get_offset(self, path):
         off = bks_.get_offset(self.state, path[0])
-        return (off or {}).get(path[1])
+        value = (off or {}).get(path[1])
+        # for activities stream, if existing state contains offset greater
+        # than 250K, then return 0, as API will raise max_skip error
+        if value and path[0] == "activities" and value > 250000:
+            return 0
+        return value
 
     def set_offset(self, path, val):
         bks_.set_offset(self.state, path[0], path[1], val)

tap_closeio/streams.py

@@ -12,6 +12,9 @@
 
 LOGGER = singer.get_logger()
 
+# default date date window size in days
+DATE_WINDOW_SIZE = 15
+
 PATHS = {
     IDS.CUSTOM_FIELDS: "/custom_fields/lead/",
     IDS.LEADS: "/lead/",
@@ -115,15 +118,24 @@ def paginated_sync(tap_stream_id, ctx, request, start_date):
             # There may be streams other than `leads` that will run into
             # `max_skip` errors but YAGNI. We can make the tap more
             # complicated once we have an extant need for it.
-            if 'max_skip = ' in str(e) and tap_stream_id == IDS.LEADS:
-                LOGGER.info(("Hit max_skip error. "
-                             "Setting bookmark to `{}` and restarting pagination.".format(
-                                 max_bookmark)))
-                skip = 0
-                ctx.clear_offsets(tap_stream_id)
-                ctx.set_bookmark(bookmark(tap_stream_id), max_bookmark)
-                _request = create_leads_request(ctx)
-                ctx.write_state()
+            if 'max_skip = ' in str(e):
+                if tap_stream_id == IDS.ACTIVITIES:
+                    LOGGER.warning("Hit max_skip error so clearing skip offset, please reduce the date window size and try again.")
+                    # clear offset
+                    ctx.clear_offsets(tap_stream_id)
+                    ctx.write_state()
+                    raise
+                elif tap_stream_id == IDS.LEADS:
+                    LOGGER.info(("Hit max_skip error. "
+                                "Setting bookmark to `{}` and restarting pagination.".format(
+                                    max_bookmark)))
+                    skip = 0
+                    ctx.clear_offsets(tap_stream_id)
+                    ctx.set_bookmark(bookmark(tap_stream_id), max_bookmark)
+                    _request = create_leads_request(ctx)
+                    ctx.write_state()
+                else:
+                    raise
             else:
                 raise
     ctx.clear_offsets(tap_stream_id)
@@ -168,15 +180,15 @@ def sync_activities(ctx):
 
     try:
         # get date window from config
-        date_window = int(ctx.config.get("date_window", 15))
-        # if date_window is 0, '0' or None, then set default window size of 15 days
+        date_window = int(ctx.config.get("date_window", DATE_WINDOW_SIZE))
+        # if date_window is 0, '0' or None, then set the default window size to DATE_WINDOW_SIZE (15 days)
         if not date_window:
-            LOGGER.warning("Invalid value of date window is passed: \'{}\', using default window size of 15 days.".format(ctx.config.get("date_window")))
-            date_window = 15
+            LOGGER.warning("Invalid value of date window is passed: \'{}\', using default window size of {} days.".format(ctx.config.get("date_window"), DATE_WINDOW_SIZE))
+            date_window = DATE_WINDOW_SIZE
     except ValueError:
-        LOGGER.warning("Invalid value of date window is passed: \'{}\', using default window size of 15 days.".format(ctx.config.get("date_window")))
+        LOGGER.warning("Invalid value of date window is passed: \'{}\', using default window size of {} days.".format(ctx.config.get("date_window"), DATE_WINDOW_SIZE))
         # In case of empty string(''), use default window
-        date_window = 15
+        date_window = DATE_WINDOW_SIZE
 
     LOGGER.info("Using offset seconds {}".format(offset_secs))
     start_date -= timedelta(seconds=offset_secs)

tests/unittests/test_activity_stream_date_window.py

@@ -14,7 +14,7 @@ def test_activity_stream_default_date_window(self, mocked_paginated_sync):
         # now date
         now_date = datetime.now()
         config = {
-            "start_date": (now_date - timedelta(days=40)).strftime("%Y-%m-%d/"), # set date 40 days later than now
+            "start_date": (now_date - timedelta(days=40)).strftime("%Y-%m-%d"), # set date 40 days later than now
             "api_key": "test_API_key"
         }
         state = {}

tests/unittests/test_activity_stream_offset.py

@@ -0,0 +1,41 @@
+from tap_closeio.schemas import IDS
+import unittest
+from tap_closeio.context import Context
+from tap_closeio.streams import paginated_sync, create_request
+
+class TestExistingStateOffset(unittest.TestCase):
+    def test_existing_state_existing_state_offset_greater_than_250K(self):
+        config = {
+            "start_date": "2022-01-01",
+            "api_key": "test_API_key"
+        }
+        state = {
+            "currently_syncing": "activities",
+            "bookmarks": {
+                "activities": {
+                    "date_created": "2022-04-01T00:00:00",
+                    "offset": {"skip": 259000}
+                }
+            }
+        }
+        context = Context(config, state)
+        offset = context.get_offset(["activities", "skip"])
+        self.assertEqual(offset, 0)
+
+    def test_existing_state_existing_state_offset_lesser_than_250K(self):
+        config = {
+            "start_date": "2022-01-01",
+            "api_key": "test_API_key"
+        }
+        state = {
+            "currently_syncing": "activities",
+            "bookmarks": {
+                "activities": {
+                    "date_created": "2022-04-01T00:00:00",
+                    "offset": {"skip": 1000}
+                }
+            }
+        }
+        context = Context(config, state)
+        offset = context.get_offset(["activities", "skip"])
+        self.assertEqual(offset, 1000)