Release/32

tap_mambu/schema.py

@@ -61,7 +61,7 @@
     'loan_accounts': {
         'key_properties': ['id'],
         'replication_method': 'INCREMENTAL',
-        'replication_keys': ['last_modified_date']
+        'replication_keys': ['last_modified_date', 'last_account_appraisal_date']
     },
     'loan_repayments': {
         'key_properties': ['encoded_key'],

tap_mambu/tap_mambu_refactor/TapProcessors/processor.py

@@ -43,18 +43,38 @@ def process_streams_from_generators(self, generators):
                     self.generator_values.pop(iterator)
             if not self.generator_values:
                 break
-            # Find lowest value in the list
+            # Find lowest value in the list, and if two or more are equal, find max bookmark
             min_record_key = None
             min_record_value = None
+            min_record_bookmark = None
             for iterator in self.generator_values:
-                if min_record_value is None \
+                bookmark_field = convert(iterator.endpoint_config.get('bookmark_field', ''))
+                # Different record
+                if min_record_key is None \
                         or min_record_value > self.generator_values[iterator][self.deduplication_key]:
                     min_record_key = iterator
                     min_record_value = self.generator_values[iterator][self.deduplication_key]
+                    if not bookmark_field:
+                        continue
+                    min_record_bookmark = self.generator_values[iterator][bookmark_field]
+                # Same record
+                elif min_record_value == self.generator_values[iterator][self.deduplication_key]:
+                    if not bookmark_field:
+                        continue
+
+                    # Check the new bookmark against the min_record_key's bookmark
+                    min_record_dttm = strptime_to_utc(min_record_bookmark)
+                    record_dttm = strptime_to_utc(self.generator_values[iterator][bookmark_field])
+                    # If min_record has bookmark smaller than record, then replace min_record (we want highest bookmark)
+                    if min_record_dttm < record_dttm:
+                        min_record_key = iterator
+                        min_record_value = self.generator_values[iterator][self.deduplication_key]
+                        min_record_bookmark = self.generator_values[iterator][bookmark_field]
 
             # Process the record
             record = self.generator_values[min_record_key]
-            if self.process_record(record, min_record_key.time_extracted):
+            if self.process_record(record, min_record_key.time_extracted,
+                                   min_record_key.endpoint_config.get('bookmark_field', '')):
                 record_count += 1
                 record_count += self._process_child_records(record)
 
@@ -70,10 +90,10 @@ def process_streams_from_generators(self, generators):
     def _process_child_records(self, record):
         return 0
 
-    def __is_record_past_bookmark(self, transformed_record):
+    def __is_record_past_bookmark(self, transformed_record, bookmark_field):
         is_record_past_bookmark = False
         bookmark_type = self.generators[0].endpoint_config.get('bookmark_type')
-        bookmark_field = convert(self.generators[0].endpoint_config.get('bookmark_field', ''))
+        bookmark_field = convert(bookmark_field)
 
         # Reset max_bookmark_value to new value if higher
         if bookmark_field and (bookmark_field in transformed_record):
@@ -101,13 +121,13 @@ def __is_record_past_bookmark(self, transformed_record):
 
         return is_record_past_bookmark
 
-    def process_record(self, record, time_extracted):
+    def process_record(self, record, time_extracted, bookmark_field):
         with Transformer() as transformer:
             transformed_record = transformer.transform(record,
                                                        self.schema,
                                                        self.stream_metadata)
 
-        if self.__is_record_past_bookmark(transformed_record):
+        if self.__is_record_past_bookmark(transformed_record, bookmark_field):
             write_record(self.stream_name,
                          transformed_record,
                          time_extracted=time_extracted)

tests/base.py

@@ -144,7 +144,7 @@ def expected_metadata(self):
                 },
                 self.REPLICATION_METHOD: "INCREMENTAL",
                 self.REPLICATION_KEYS: {
-                    "last_modified_date"
+                    "last_modified_date", "last_account_appraisal_date"
                 }
             },
             "loan_repayments": {

tests/test_bookmarks.py

@@ -82,7 +82,7 @@ def test_run(self):
                 second_sync_messages = second_sync_records.get(stream, {}).get('messages', [])
 
                 if replication_method == self.INCREMENTAL:
-                    replication_key = self.expected_replication_keys().get(stream).pop()
+                    replication_keys = self.expected_replication_keys().get(stream)
 
                     first_sync_bookmark_value = first_sync_bookmarks['bookmarks'][stream]
                     second_sync_bookmark_value = second_sync_bookmarks['bookmarks'][stream]
@@ -92,29 +92,38 @@ def test_run(self):
                     self.assertEqual(first_sync_bookmark_value,
                                      second_sync_bookmark_value)
 
-                    # Verify that first sync records fall betwen the start date and the final
+                    # Verify that first sync records fall between the start date and the final
                     # bookmark value
                     for message in first_sync_messages:
                         lower_bound = strptime_to_utc(self.get_properties()['start_date'])
-                        actual_value = strptime_to_utc(message.get('data').get(replication_key))
                         upper_bound = strptime_to_utc(first_sync_bookmark_value)
+                        record = message.get('data')
+                        actual_values = [strptime_to_utc(record.get(replication_key))
+                                         for replication_key in replication_keys if replication_key in record]
+
+                        self.assertNotEqual(actual_values, [], msg="No replication key found in record")
+
                         self.assertTrue(
-                            lower_bound <= actual_value <= upper_bound,
+                            any([lower_bound <= actual_value <= upper_bound for actual_value in actual_values]),
                             msg="First sync records fall outside of expected sync window"
                         )
 
                     # Verify the second sync records fall between simulated bookmark value and the
                     # final bookmark value
                     for message in second_sync_messages:
                         lower_bound = strptime_to_utc(simulated_bookmark_value)
-                        actual_value = strptime_to_utc(message.get('data',{}).get(replication_key))
                         upper_bound = strptime_to_utc(second_sync_bookmark_value)
+                        record = message.get('data')
+                        actual_values = [strptime_to_utc(record.get(replication_key))
+                                         for replication_key in replication_keys if replication_key in record]
+
+                        self.assertNotEqual(actual_values, [], msg="No replication key found in record")
+
                         self.assertTrue(
-                            lower_bound <= actual_value <= upper_bound,
+                            any([lower_bound <= actual_value <= upper_bound for actual_value in actual_values]),
                             msg="Second sync records fall outside of expected sync window"
                         )
 
-
                     # Verify the number of records in the 2nd sync is less then the first
                     self.assertLess(second_sync_count, first_sync_count)
 

tests/test_start_date.py

@@ -37,7 +37,7 @@ def get_replication_key_values(self, stream, records):
         for record in records:
             # Build the primary key for this record, maintaining the same order for the fields
             record_rep_key = [record[field]
-                              for field in sorted(self.expected_replication_keys()[stream])]
+                              for field in sorted(self.expected_replication_keys()[stream]) if field in record]
             # Cast to a tuple to make it hashable
             all_records.append(record_rep_key[0])
         return all_records