Add intermediate bookmarking implementation

tap_activecampaign/streams.py

@@ -38,7 +38,8 @@ class ActiveCampaign:
     bookmark_query_field = None
     links = []
     children = []
-
+    offset_bookmark_limit = 1000
+
     def __init__(self, client: ActiveCampaignClient = None):
         self.client = client
 
@@ -76,24 +77,30 @@ def write_record(self, stream_name, record, time_extracted):
             raise err
 
     def get_bookmark(self, state, stream, default):
-        """ 
+        """
         Return bookmark value present in state or return a default value if no bookmark
         present in the state for provided stream
         """
         if (state is None) or ('bookmarks' not in state):
-            return default
-        return (
-            state
-            .get('bookmarks', {})
-            .get(stream, default)
-        )
+            return default, 0
 
+        if next(iter(self.replication_keys or []), None):
+            return state.get('bookmarks', {}).get(stream, default), 0
 
-    def write_bookmark(self, state, stream, value):
+        return default, state.get('bookmarks', {}).get(stream, 0)
+
+
+    def write_bookmark(self, state, stream, value, offset=None):
         """ Write bookmark in state. """
         if 'bookmarks' not in state:
             state['bookmarks'] = {}
-        state['bookmarks'][stream] = value
+        if next(iter(self.replication_keys or []), None):
+            state['bookmarks'][stream] = value
+        elif offset:
+            state['bookmarks'][stream] = offset
+        elif stream in state['bookmarks']:
+            del state['bookmarks'][stream]
+
         LOGGER.info('Write state for stream: {}, value: {}'.format(stream, value))
         singer.write_state(state)
 
@@ -190,22 +197,19 @@ def sync(
         last_datetime = None
         max_bookmark_value = None
 
-        last_datetime = self.get_bookmark(state, self.stream_name, start_date)
+        last_datetime, offset = self.get_bookmark(state, self.stream_name, start_date)
         max_bookmark_value = last_datetime
-        LOGGER.info('stream: {}, bookmark_field: {}, last_datetime: {}'.format(
-            self.stream_name, bookmark_field, last_datetime))
-        now_datetime = utils.now()
-        last_dttm = strptime_to_utc(last_datetime)
+        LOGGER.info(
+            f'stream: {self.stream_name}, bookmark_field: {bookmark_field}, last_datetime: {last_datetime}, offset: {offset}')
         endpoint_total = 0
 
         # pagination: loop thru all pages of data
         # Pagination reference: https://developers.activecampaign.com/reference#pagination
         # Each page has an offset (starting value) and a limit (batch size, number of records)
         # Increase the "offset" by the "limit" for each batch.
-        # Continue until the "record_count" returned < "limit" is null/zero or 
-        offset = 0 # Starting offset value for each batch API call
+        # Continue until the "record_count" returned < "limit" is null/zero
         limit = 100 # Batch size; Number of records per API call; Max = 100
-        total_records = 0 # Initialize total
+        total_records = offset # Initialize total
         record_count = limit # Initialize, reset for each API call
         page = 1
 
@@ -221,8 +225,8 @@ def sync(
 
             # Need URL querystring for 1st page; subsequent pages provided by next_url
             # querystring: Squash query params into string
-            querystring = None
-            querystring = '&'.join(['%s=%s' % (key, value) for (key, value) in params.items()])
+            querystring = f'orders[{bookmark_field}]=ASC&' if bookmark_field else ""
+            querystring += '&'.join(['%s=%s' % (key, value) for (key, value) in params.items()])
 
             LOGGER.info('URL for Stream {}: {}{}{}'.format(
                 self.stream_name,
@@ -235,10 +239,12 @@ def sync(
                                 querystring, path, max_bookmark_value, state, catalog, start_date, last_datetime, endpoint_total, 
                                   limit, total_records, record_count, page, offset, parent, parent_id, selected_streams)
 
+            # Write offset after every 1000 records
+            if not offset % self.offset_bookmark_limit:
+                self.write_bookmark(state, self.stream_name, max_bookmark_value, offset)
+
         # Update the state with the max_bookmark_value for the endpoint
-        # ActiveCampaign API does not allow page/batch sorting; bookmark written for endpoint
-        if bookmark_field:
-            self.write_bookmark(state, self.stream_name, max_bookmark_value)
+        self.write_bookmark(state, self.stream_name, max_bookmark_value)
 
         # Return total_records (for all pages and date windows)
         return endpoint_total
@@ -287,6 +293,7 @@ def sync_child_stream(self, children, transformed_data, catalog, state, start_da
                 for record in transformed_data:
                     i = 0
                     # Set parent_id
+                    parent_id_field = None
                     for id_field in id_fields:
                         if i == 0:
                             parent_id_field = id_field

tests/base.py

@@ -548,12 +548,16 @@ def select_all_streams_and_fields(conn_id, catalogs, select_all_fields: bool = T
             connections.select_catalog_and_fields_via_metadata(
                 conn_id, catalog, schema, [], non_selected_properties)
 
-    def calculated_states_by_stream(self, current_state):
+    def calculated_states_by_stream(self, current_state, currently_syncing_stream=None):
         timedelta_by_stream = {stream: [0,0,1]  # {stream_name: [days, hours, minutes], ...}
                                for stream in self.expected_check_streams()}
 
         stream_to_calculated_state = {stream: "" for stream in current_state['bookmarks'].keys()}
         for stream, state in current_state['bookmarks'].items():
+            # Skip the currently syncing stream
+            if (currently_syncing_stream and stream == currently_syncing_stream) or self.expected_replication_method()[stream] == 'FULL_TABLE':
+                continue
+
             state_as_datetime = dateutil.parser.parse(state)
 
             days, hours, minutes = timedelta_by_stream[stream]

tests/test_interrupted_sync.py

@@ -13,7 +13,20 @@ def name(self):
     def test_name(self):
         LOGGER.info("Interrupted Sync test for tap-activecampaign")
 
-    def test_run(self):
+    def test_run_incremental_streams(self):
+        streams_to_test = self.expected_check_streams() - {'campaign_lists', 'contact_conversions', 'sms'}
+        currently_syncing_stream = 'activities'
+        bookmark_field = list(self.expected_replication_keys()[currently_syncing_stream])[0]
+        interrupted_stream_bookmark = '2021-12-01T00:00:00.000000Z'
+        self.main(streams_to_test, currently_syncing_stream, interrupted_stream_bookmark)
+
+    def test_run_full_table_streams(self):
+        streams_to_test = {'accounts', 'activities', 'campaign_lists', 'tags'}
+        currently_syncing_stream = 'campaign_lists'
+        interrupted_stream_bookmark = 10000
+        self.main(streams_to_test, currently_syncing_stream, interrupted_stream_bookmark)
+
+    def main(self, streams_to_test, currently_syncing_stream, interrupted_stream_bookmark):
         """
         Scenario: A sync job is interrupted. The state is saved with `currently_syncing`.
                   The next sync job kicks off and the tap picks back up on that
@@ -47,9 +60,6 @@ def test_run(self):
 
         conn_id = connections.ensure_connection(self)
 
-        # Note: test data not available for following streams: contact_conversions, sms
-        streams_to_test = self.expected_check_streams() - {'contact_conversions', 'sms'}
-
         # Run check mode
         found_catalogs = self.run_and_verify_check_mode(conn_id)
 
@@ -81,10 +91,13 @@ def test_run(self):
 
         interrupted_sync_state = {'bookmarks': dict()}
         simulated_states = self.calculated_states_by_stream(
-            first_sync_bookmarks)
+            first_sync_bookmarks, currently_syncing_stream)
         for stream, new_state in simulated_states.items():
             interrupted_sync_state['bookmarks'][stream] = new_state
-        interrupted_sync_state["currently_syncing"] = "messages"
+
+        interrupted_sync_state['currently_syncing'] = currently_syncing_stream
+        interrupted_sync_state['bookmarks'][currently_syncing_stream] = interrupted_stream_bookmark
+
         menagerie.set_state(conn_id, interrupted_sync_state)
 
         ##########################################################################
@@ -113,6 +126,10 @@ def test_run(self):
             self.assertDictEqual(post_interrupted_sync_state, post_interrupted_sync_state,
                                  msg="Final state after interruption should be equal to full sync")
 
+            # Verify final_state doesn't have offset key
+            self.assertNotIn("offset", post_interrupted_sync_state,
+                             msg="Final state after interruption should be equal to full sync")
+
         # Stream level assertions
         for stream in streams_to_test:
             with self.subTest(stream=stream):
@@ -136,53 +153,57 @@ def test_run(self):
                 full_sync_record_count = len(first_sync_stream_records)
                 interrupted_record_count = len(post_interrupted_sync_stream_records)
 
+                # Final bookmark after interrupted sync
+                final_stream_bookmark = post_interrupted_sync_state["bookmarks"].get(
+                    stream, None)
                 if replication_method == self.INCREMENTAL:
-                    # Final bookmark after interrupted sync
-                    final_stream_bookmark = post_interrupted_sync_state["bookmarks"].get(
-                        stream, None)
-
                     # Verify final bookmark matched the formatting standards for the resuming sync
                     self.assertIsNotNone(final_stream_bookmark,
                                          msg="Bookmark can not be 'None'.")
                     self.assertIsInstance(final_stream_bookmark, str,
                                           msg="Bookmark format is not as expected.")
+                else:
+                    self.assertNotIn(stream, post_interrupted_sync_state)
 
                 if stream == interrupted_sync_state["currently_syncing"]:
-                    # Assign the start date to the interrupted stream
-                    interrupted_stream_datetime = self.parse_date(
-                        interrupted_sync_state["bookmarks"][stream])
+                    if self.expected_replication_method()[stream] == self.INCREMENTAL:
+                        # Assign the start date to the interrupted stream
+                        interrupted_stream_datetime = self.parse_date(
+                            interrupted_sync_state["bookmarks"][stream])
+
+                        primary_key = self.expected_primary_keys()[stream].pop() if self.expected_primary_keys()[stream] else None
+
+                        # Get primary keys of 1st sync records
+                        if primary_key:
+                            full_records_primary_keys = [x.get(primary_key)
+                                                        for x in first_sync_stream_records]
 
-                    primary_key = self.expected_primary_keys()[stream].pop() if self.expected_primary_keys()[stream] else None
+                        for record in post_interrupted_sync_stream_records:
+                            record_time = self.parse_date(record.get(replication_key))
 
-                    # Get primary keys of 1st sync records
-                    if primary_key:
-                        full_records_primary_keys = [x.get(primary_key)
-                                                     for x in first_sync_stream_records]
+                            # Verify resuming sync only replicates records with the replication key
+                            # values greater or equal to the state for streams that were replicated
+                            # during the interrupted sync.
+                            self.assertGreaterEqual(record_time, interrupted_stream_datetime)
 
-                    for record in post_interrupted_sync_stream_records:
-                        record_time = self.parse_date(record.get(replication_key))
+                            # Verify the interrupted sync replicates the expected record set all
+                            # interrupted records are in full records
+                            if primary_key:
+                                self.assertIn(record[primary_key], full_records_primary_keys,
+                                            msg="Incremental table record in interrupted sync not found in full sync")
 
-                        # Verify resuming sync only replicates records with the replication key
-                        # values greater or equal to the state for streams that were replicated
-                        # during the interrupted sync.
-                        self.assertGreaterEqual(record_time, interrupted_stream_datetime)
+                        # Record count for all streams of interrupted sync match expectations
+                        records_after_interrupted_bookmark = 0
+                        for record in first_sync_stream_records:
+                            record_time = self.parse_date(record.get(replication_key))
+                            if record_time >= interrupted_stream_datetime:
+                                records_after_interrupted_bookmark += 1
 
-                        # Verify the interrupted sync replicates the expected record set all
-                        # interrupted records are in full records
-                        if primary_key:
-                            self.assertIn(record[primary_key], full_records_primary_keys,
-                                          msg="Incremental table record in interrupted sync not found in full sync")
-
-                    # Record count for all streams of interrupted sync match expectations
-                    records_after_interrupted_bookmark = 0
-                    for record in first_sync_stream_records:
-                        record_time = self.parse_date(record.get(replication_key))
-                        if record_time >= interrupted_stream_datetime:
-                            records_after_interrupted_bookmark += 1
-
-                    self.assertEqual(records_after_interrupted_bookmark, interrupted_record_count,
-                                     msg="Expected {} records in each sync".format(
-                                         records_after_interrupted_bookmark))
+                        self.assertEqual(records_after_interrupted_bookmark, interrupted_record_count,
+                                        msg="Expected {} records in each sync".format(
+                                            records_after_interrupted_bookmark))
+                    else:
+                        self.assertGreater(first_sync_record_count[stream], interrupted_record_count)
 
                 else:
                     # Get the date to start 2nd sync for non-interrupted streams