[TDL-22683] Use Date Windowing Pagination

.circleci/config.yml

@@ -1,8 +1,20 @@
-version: 2
-jobs:
-  build:
+version: 2.1
+orbs:
+  slack: circleci/[email protected]
+
+executors:
+  docker-executor:
     docker:
       - image: 218546966473.dkr.ecr.us-east-1.amazonaws.com/circle-ci:stitch-tap-tester
+
+jobs:
+  build:
+    executor: docker-executor
+    steps:
+      - run: echo "CI Done"
+
+  ensure_env:
+    executor: docker-executor
     steps:
       - checkout
       - run:
@@ -12,39 +24,93 @@ jobs:
             source /usr/local/share/virtualenvs/tap-mambu/bin/activate
             pip install -U 'pip<19.2' 'setuptools<51.0.0'
             pip install .[dev]
-      - add_ssh_keys
-      - run:
-          name: 'JSON Validator'
-          command: |
-            source /usr/local/share/virtualenvs/tap-tester/bin/activate
-            stitch-validate-json tap_mambu/helpers/schemas/*.json
+            aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/tap_tester_sandbox /usr/local/share/virtualenvs/dev_env.sh
+      - slack/notify-on-failure:
+          only_for_branches: master
+      - persist_to_workspace:
+          root: /usr/local/share/virtualenvs
+          paths:
+            - tap-mambu
+            - dev_env.sh
+
+  run_unit_tests:
+    executor: docker-executor
+    steps:
+      - checkout
+      - attach_workspace:
+          at: /usr/local/share/virtualenvs
       - run:
-          when: always
           name: 'Unit Tests'
           command: |
             source /usr/local/share/virtualenvs/tap-mambu/bin/activate
+            pip install parameterized
             pytest tests/unittests
+      - store_test_results:
+          path: test_output/report.xml
+      - store_artifacts:
+          path: htmlcov
+
+  run_integration_tests:
+    executor: docker-executor
+    parallelism: 5
+    steps:
+      - checkout
+      - attach_workspace:
+          at: /usr/local/share/virtualenvs
       - run:
-          name: 'Integration Tests'
+          name: 'Run Integration Tests'
+          no_output_timeout: 30m
           command: |
-            aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/tap_tester_sandbox dev_env.sh
-            source dev_env.sh
+            source /usr/local/share/virtualenvs/dev_env.sh
+            mkdir /tmp/${CIRCLE_PROJECT_REPONAME}
+            export STITCH_CONFIG_DIR=/tmp/${CIRCLE_PROJECT_REPONAME}
             source /usr/local/share/virtualenvs/tap-tester/bin/activate
-            run-test --tap=tap-mambu tests
+            circleci tests glob "tests/test_*.py" | circleci tests split > ./tests-to-run
+            if [ -s ./tests-to-run ]; then
+              for test_file in $(cat ./tests-to-run)
+              do
+                echo $test_file > $STITCH_CONFIG_DIR/tap_test.txt
+                run-test --tap=${CIRCLE_PROJECT_REPONAME} $test_file
+              done
+            fi
+      - slack/notify-on-failure:
+          only_for_branches: master
+      - store_artifacts:
+          path: /tmp/tap-mambu
+
 workflows:
   version: 2
-  commit:
+  commit: &commit_jobs
     jobs:
+      - ensure_env:
+          context:
+            - circleci-user
+            - tier-1-tap-user
+      - run_unit_tests:
+          context:
+            - circleci-user
+            - tier-1-tap-user
+          requires:
+            - ensure_env
+      - run_integration_tests:
+          context:
+            - circleci-user
+            - tier-1-tap-user
+          requires:
+            - ensure_env
       - build:
-          context: circleci-user
+          context:
+            - circleci-user
+            - tier-1-tap-user
+          requires:
+            - run_unit_tests
+            - run_integration_tests
   build_daily:
+    <<: *commit_jobs
     triggers:
       - schedule:
-          cron: "0 0 * * *"
+          cron: "0 1 * * *"
           filters:
             branches:
               only:
                 - master
-    jobs:
-      - build:
-          context: circleci-user

README.md

@@ -254,10 +254,13 @@ This tap:
         "lookback_window": 30,
         "user_agent": "tap-mambu <api_user_email@your_company.com>",
         "page_size": "500",
-        "apikey_audit": "AUDIT_TRAIL_APIKEY"
+        "apikey_audit": "AUDIT_TRAIL_APIKEY",
+        "window_size": 7
     }
     ```
-
+
+    Note: The `window_size` parameter defaults to 1 day, which may cause slowdowns in historical sync for streams utilizing multi-threaded implementation. Conversely, using a larger `window_size` could lead to potential `out-of-memory` issues. It is advisable to select an optimal `window_size` based on the `start_date` and volume of data to mitigate these concerns.
+
     Optionally, also create a `state.json` file. `currently_syncing` is an optional attribute used for identifying the last object to be synced in case the job is interrupted mid-stream. The next run would begin where the last job left off.
 
     ```json

tap_mambu/__init__.py

@@ -35,7 +35,8 @@ def main():
                      parsed_args.config['subdomain'],
                      parsed_args.config.get('apikey_audit'),
                      int(parsed_args.config.get('page_size', DEFAULT_PAGE_SIZE)),
-                     user_agent=parsed_args.config['user_agent']) as client:
+                     user_agent=parsed_args.config['user_agent'],
+                     window_size=parsed_args.config.get('window_size')) as client:
 
         state = {}
         if parsed_args.state:

tap_mambu/helpers/client.py

@@ -1,9 +1,11 @@
 import backoff
 import requests
 import requests.adapters
-from requests.exceptions import ConnectionError
+from requests.exceptions import ConnectionError, ChunkedEncodingError
 from singer import metrics, get_logger
 
+from urllib3.exceptions import ProtocolError
+from tap_mambu.helpers.constants import DEFAULT_DATE_WINDOW_SIZE
 
 LOGGER = get_logger()
 class ClientError(Exception):
@@ -118,13 +120,21 @@ def __init__(self,
                  subdomain,
                  apikey_audit,
                  page_size,
-                 user_agent=''):
+                 user_agent='',
+                 window_size=DEFAULT_DATE_WINDOW_SIZE):
         self.__username = username
         self.__password = password
         self.__subdomain = subdomain
         base_url = "https://{}.mambu.com/api".format(subdomain)
         self.base_url = base_url
         self.page_size = page_size
+        try:
+            self.window_size = int(float(window_size)) if window_size else DEFAULT_DATE_WINDOW_SIZE
+            if self.window_size <= 0:
+                raise ValueError()
+        except ValueError:
+            raise Exception("The entered window size '{}' is invalid; it should be a valid non-zero integer.".format(window_size))
+
         self.__user_agent = f'MambuTap-{user_agent}' if user_agent else 'MambuTap'
         self.__apikey = apikey
         self.__session = requests.Session()
@@ -178,7 +188,8 @@ def check_access(self):
             return True
 
     @backoff.on_exception(backoff.expo,
-                          (MambuInternalServiceError, ConnectionError, MambuApiLimitError),
+                          (MambuInternalServiceError, ConnectionError,
+                           ChunkedEncodingError, MambuApiLimitError, ProtocolError),
                           max_tries=7,
                           factor=3)
     def request(self, method, path=None, url=None, json=None, version=None, apikey_type=None, **kwargs):

tap_mambu/helpers/constants.py

@@ -1 +1,2 @@
 DEFAULT_PAGE_SIZE = 200
+DEFAULT_DATE_WINDOW_SIZE = 1

tap_mambu/helpers/exceptions.py

@@ -3,3 +3,6 @@ class NoDeduplicationCapabilityException(Exception):
 
 class NoDeduplicationKeyException(Exception):
     pass
+
+class MambuGeneratorThreadNotAlive(Exception):
+    pass

tap_mambu/helpers/multithreaded_requests.py

@@ -1,7 +1,6 @@
 import singer
 from concurrent.futures import Future, ThreadPoolExecutor
 from typing import List
-from .perf_metrics import PerformanceMetrics
 
 
 LOGGER = singer.get_logger()
@@ -24,16 +23,15 @@ def run(client, stream_name,
                     f'{endpoint_api_version}): {client.base_url}/{endpoint_path}?{endpoint_querystring}'
                     f' - body = {endpoint_body}')
 
-        with PerformanceMetrics(metric_name="generator"):
-            response = client.request(
-                method=endpoint_api_method,
-                path=endpoint_path,
-                version=endpoint_api_version,
-                apikey_type=endpoint_api_key_type,
-                params=endpoint_querystring,
-                endpoint=stream_name,
-                json=endpoint_body
-            )
+        response = client.request(
+            method=endpoint_api_method,
+            path=endpoint_path,
+            version=endpoint_api_version,
+            apikey_type=endpoint_api_key_type,
+            params=endpoint_querystring,
+            endpoint=stream_name,
+            json=endpoint_body
+        )
 
         LOGGER.info(f'(generator) Stream {stream_name} - extracted records: {len(response)}')
         return response