Changes:

1) Removed endpoint params from activities generator.
2) Overrode modify_reques_params method for Loan Transactions generator.
3) Removed the method _all_fetch_batch_steps from multithreaded bookmark generator.
4) Modified the implementation for _all_fetch_batch_steps method in multithreaded offset generator to include date windowing.
5) Added new method modify_reques_params in multithreaded offset generator.

tap_mambu/tap_generators/activities_generator.py

@@ -13,10 +13,6 @@ def _init_endpoint_config(self):
         self.endpoint_path = "activities"
         self.endpoint_api_method = "GET"
         self.endpoint_api_version = "v1"
-
-        self.endpoint_params["from"] = datetime_to_utc_str(str_to_localized_datetime(
-                    get_bookmark(self.state, self.stream_name, self.sub_type, self.start_date)))[:10]
-        self.endpoint_params["to"] = datetime_to_utc_str(utc_now())[:10]
         self.endpoint_bookmark_field = "timestamp"
 
     @staticmethod

tap_mambu/tap_generators/loan_transactions_generator.py

@@ -1,20 +1,30 @@
 from .multithreaded_bookmark_generator import MultithreadedBookmarkGenerator
-from ..helpers import get_bookmark
-from ..helpers.datetime_utils import datetime_to_utc_str, str_to_localized_datetime
+from ..helpers.datetime_utils import datetime_to_utc_str
+from datetime import datetime
 
 
 class LoanTransactionsGenerator(MultithreadedBookmarkGenerator):
+    def __init__(self, stream_name, client, config, state, sub_type):
+        super(LoanTransactionsGenerator, self).__init__(stream_name, client, config, state, sub_type)
+        self.date_windowing = True
+
     def _init_endpoint_config(self):
         super(LoanTransactionsGenerator, self)._init_endpoint_config()
         self.endpoint_path = "loans/transactions:search"
         self.endpoint_bookmark_field = "creationDate"
         self.endpoint_sorting_criteria["field"] = "id"
-        self.endpoint_filter_criteria = [
+
+    def modify_request_params(self, start, end):
+        self.endpoint_body['filterCriteria'] = [
             {
                 "field": "creationDate",
                 "operator": "AFTER",
-                "value": datetime_to_utc_str(str_to_localized_datetime(
-                    get_bookmark(self.state, self.stream_name, self.sub_type, self.start_date)))
+                "value": datetime.strftime(start, '%Y-%m-%dT00:00:00.000000Z')
+            },
+            {
+                "field": "creationDate",
+                "operator": "BEFORE",
+                "value": datetime.strftime(end, '%Y-%m-%dT23:59:59.000000Z')
             }
         ]
 

tap_mambu/tap_generators/multithreaded_bookmark_generator.py

@@ -1,10 +1,8 @@
 import datetime
 import time
-import backoff
 
 from copy import deepcopy
 from singer import get_logger
-from datetime import datetime, timedelta
 
 from .multithreaded_offset_generator import MultithreadedOffsetGenerator
 from ..helpers import transform_json, convert
@@ -53,32 +51,6 @@ def queue_batches(self):
                                                                    deepcopy(self.params)))
         return futures
 
-    @backoff.on_exception(backoff.expo, RuntimeError, max_tries=5)
-    def _all_fetch_batch_steps(self):
-        if self.date_windowing:
-            start = datetime.strptime(self.params["from"], '%Y-%m-%d').date()
-            end = datetime.strptime(self.params["to"], '%Y-%m-%d').date()
-            temp = start + timedelta(days=self.date_window_size)
-            stop_iteration = True
-            while temp < end:
-                if stop_iteration:
-                    self.offset = 0
-                self.static_params["from"] = datetime.strftime(start, '%Y-%m-%d')
-                self.static_params["to"] = datetime.strftime(temp, '%Y-%m-%d')
-                final_buffer, stop_iteration = self.collect_batches(self.queue_batches())
-                self.preprocess_batches(final_buffer)
-                if not final_buffer or stop_iteration:
-                    start = temp
-                    temp = start + timedelta(days=self.date_window_size)
-            self.offset = 0
-            self.static_params["from"] = datetime.strftime(start, '%Y-%m-%d')
-            self.static_params["to"] = datetime.strftime(end, '%Y-%m-%d')
-        final_buffer, stop_iteration = self.collect_batches(self.queue_batches())
-        self.preprocess_batches(final_buffer)
-        if not final_buffer or stop_iteration:
-            return False
-        return True
-
     def collect_batches(self, futures):
         # wait for responses, and check them for errors
         final_buffer = set()

tap_mambu/tap_generators/multithreaded_offset_generator.py

@@ -4,9 +4,11 @@
 
 import backoff
 from singer import get_logger
+from datetime import datetime, timedelta
 
 from .generator import TapGenerator
-from ..helpers import transform_json
+from ..helpers import transform_json, get_bookmark
+from ..helpers.datetime_utils import str_to_localized_datetime, datetime_to_utc_str, utc_now
 from ..helpers.multithreaded_requests import MultithreadedRequestsPool
 from ..helpers.perf_metrics import PerformanceMetrics
 
@@ -120,13 +122,35 @@ def preprocess_batches(self, final_buffer):
 
     @backoff.on_exception(backoff.expo, RuntimeError, max_tries=5)
     def _all_fetch_batch_steps(self):
-        futures = self.queue_batches()
-        final_buffer, stop_iteration = self.collect_batches(futures)
+        if self.date_windowing:
+            start_datetime = datetime_to_utc_str(str_to_localized_datetime(
+                get_bookmark(self.state, self.stream_name, self.sub_type, self.start_date)))[:10]
+            end_datetime = datetime_to_utc_str(utc_now())[:10]
+            start = datetime.strptime(start_datetime, '%Y-%m-%d').date()
+            end = datetime.strptime(end_datetime, '%Y-%m-%d').date()
+            temp = start + timedelta(days=self.date_window_size)
+            stop_iteration = True
+            while temp < end:
+                if stop_iteration:
+                    self.offset = 0
+                self.modify_request_params(start, temp)
+                final_buffer, stop_iteration = self.collect_batches(self.queue_batches())
+                self.preprocess_batches(final_buffer)
+                if not final_buffer or stop_iteration:
+                    start = temp
+                    temp = start + timedelta(days=self.date_window_size)
+            self.offset = 0
+            self.modify_request_params(start, end)
+        final_buffer, stop_iteration = self.collect_batches(self.queue_batches())
         self.preprocess_batches(final_buffer)
         if not final_buffer or stop_iteration:
             return False
         return True
 
+    def modify_request_params(self, start, end):
+        self.static_params["from"] = datetime.strftime(start, '%Y-%m-%d')
+        self.static_params["to"] = datetime.strftime(end, '%Y-%m-%d')
+
     def error_check_and_fix(self, final_buffer: set, temp_buffer: set, futures: list):
         try:
             final_buffer = self.check_and_get_set_reunion(final_buffer, temp_buffer, self.artificial_limit)