Library upgrades, Extraction failures handling. (#11)

Jaideep-Qlik · Rahul Gogawale · RushiT0122 · web-flow · commit 85603c70c889 · 2025-11-24T12:24:33.000+05:30
* Libraries updated.

* Timeout and testcases added for timeout issue.

* Added unit test &amp; fixed connection error.

* Added circleci changes.

* As per PR comments required changes are done.

* Changes added for test case.

* Update README.md

Co-authored-by: Rushikesh Todkar &lt;98420315+RushiT0122@users.noreply.github.com&gt;

* Test case added in parameterized format.

* Requested changes done.

* Test cases updated.

* Added changes as per the comments.

* Fixing build..

* Updated test client for build failuer.

* Adjusted path in circle ci.

* Fixing Build...

* Code refeactor.

* build fix...

* Test updated.

* Test removed.

* pip command fix

---------

Co-authored-by: Rahul Gogawale &lt;rahul.gogawale@globallogic.com&gt;
Co-authored-by: Rushikesh Todkar &lt;98420315+RushiT0122@users.noreply.github.com&gt;
Co-authored-by: Rutuja Deshmukh &lt;107538720+rdeshmukh15@users.noreply.github.com&gt;
Co-authored-by: Rutuja Deshmukh &lt;rutuja.deshmukh@qlik.com&gt;
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -11,16 +11,43 @@ jobs:
             uv venv --python 3.9 /usr/local/share/virtualenvs/tap-mailshake
             source /usr/local/share/virtualenvs/tap-mailshake/bin/activate
             uv pip install .
+      - run:
+          name: 'JSON Validator'
+          command: |
+            source /usr/local/share/virtualenvs/tap-tester/bin/activate
+            stitch-validate-json tap_mailshake/schemas/*.json
       - run:
           name: 'pylint'
           command: |
             source /usr/local/share/virtualenvs/tap-mailshake/bin/activate
             uv pip install pylint
-            pylint tap_mailshake --disable missing-docstring,logging-format-interpolation,no-member,broad-except,too-many-branches,consider-using-f-string,raise-missing-from,broad-exception-raised,too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-nested-blocks,too-many-statements
-
+            pylint tap_mailshake -d C,R,W
+      - add_ssh_keys
+      - run:
+          name: 'Unit Tests'
+          command: |
+            source /usr/local/share/virtualenvs/tap-mailshake/bin/activate
+            uv pip install nose coverage parameterized
+            nosetests --with-coverage --cover-erase --cover-package=tap_mailshake --cover-html-dir=htmlcov tap_mailshake/tests/unittests
+            coverage html
+      - store_test_results:
+          path: test_output/report.xml
+      - store_artifacts:
+          path: htmlcov
 workflows:
   version: 2
   commit:
     jobs:
       - build:
           context: circleci-user
+  build_daily:
+    triggers:
+      - schedule:
+          cron: "0 0 * * *"
+          filters:
+            branches:
+              only:
+                - master
+    jobs:
+      - build:
+          context: circleci-user
diff --git a/README.md b/README.md
@@ -133,12 +133,13 @@ This tap uses [simple authentication](https://api-docs.mailshake.com/#Simple) as
         "api_key": "YourApiKeyFromMailshake",
         "domain": "api.mailshake.com",
         "start_date": "2019-01-01T00:00:00Z",
-        "user_agent": "tap-mailshake <api_user_email@your_company.com>"
+        "user_agent": "tap-mailshake <api_user_email@your_company.com>",
+		"request_timeout": 300
     }
     ```
-    
+    **request_timeout (optional)**: Timeout in seconds for API requests. Default request_timeout is 300.
     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.
-
+    Reference link for API documentation: [Mailshake API Docs](https://api-docs.mailshake.com/)
     ```json
     {
         "currently_syncing": "campaigns",
diff --git a/setup.py b/setup.py
@@ -9,9 +9,9 @@
       classifiers=['Programming Language :: Python :: 3 :: Only'],
       py_modules=['tap_mailshake'],
       install_requires=[
-          'backoff==1.10.0',
-          'requests==2.32.4',
-          'singer-python==5.13.2'
+          'backoff==2.2.1',
+          'requests==2.32.5',
+          'singer-python==6.1.1'
       ],
       entry_points='''
           [console_scripts]
diff --git a/tap_mailshake/__init__.py b/tap_mailshake/__init__.py
@@ -28,9 +28,13 @@ def do_discover():
 @singer.utils.handle_top_exception(LOGGER)
 def main():
     parsed_args = singer.utils.parse_args(REQUIRED_CONFIG_KEYS)
+    request_timeout = parsed_args.config.get('request_timeout')
 
-    with MailshakeClient(parsed_args.config['api_key'],
-                         parsed_args.config['user_agent']) as client:
+    with MailshakeClient(
+            parsed_args.config['api_key'],
+            parsed_args.config['user_agent'],
+            request_timeout=request_timeout
+    ) as client:
 
         state = {}
         if parsed_args.state:
diff --git a/tap_mailshake/client.py b/tap_mailshake/client.py
@@ -3,9 +3,13 @@
 import singer
 from singer import metrics, utils
 from requests.auth import HTTPBasicAuth
+from requests.exceptions import ConnectionError, ChunkedEncodingError
+import http.client
+from requests.exceptions import Timeout
 
 LOGGER = singer.get_logger()
-API_VERSION = '2017-04-01'
+API_VERSION = "2017-04-01"
+REQUEST_TIMEOUT = 300
 
 
 class Server5xxError(Exception):
@@ -103,7 +107,7 @@ def get_exception_for_error_code(error_code):
 
 
 def raise_for_error(response):
-    LOGGER.error('ERROR {}: {}, REASON: {}'.format(response.status_code,
+    LOGGER.error("ERROR {}: {}, REASON: {}".format(response.status_code,
                                                    response.text, response.reason))
     try:
         response.raise_for_status()
@@ -115,10 +119,10 @@ def raise_for_error(response):
                 # us a 2xx response nor a response content.
                 return
             response = response.json()
-            if ('error' in response) or ('errorCode' in response):
-                message = '%s: %s' % (response.get('error', str(error)),
-                                      response.get('message', 'Unknown Error'))
-                error_code = response.get('code')
+            if ("error" in response) or ("errorCode" in response):
+                message = "%s: %s" % (response.get("error", str(error)),
+                                      response.get("message", "Unknown Error"))
+                error_code = response.get("code")
                 ex = get_exception_for_error_code(error_code)
                 raise ex(message)
             raise MailshakeError(error)
@@ -134,12 +138,15 @@ class MailshakeClient:
 
     def __init__(self,
                  api_key,
-                 user_agent=None):
+                 user_agent=None,
+                 request_timeout=REQUEST_TIMEOUT):
         self.__api_key = api_key
         self.base_url = "https://api.mailshake.com/{}".format(
             API_VERSION)
         self.__user_agent = user_agent
         self.__session = requests.Session()
+        self.__verified = False
+        self.request_timeout = request_timeout or REQUEST_TIMEOUT
 
     def __enter__(self):
         self.check_access()
@@ -155,25 +162,31 @@ def __exit__(self, exception_type, exception_value, traceback):
     @utils.ratelimit(1, 1.2)
     def check_access(self):
         if self.__api_key is None:
-            raise Exception('Error: Missing api_key in tap_config.json.')
+            raise Exception("Error: Missing api_key in tap_config.json.")
         headers = {}
-        endpoint = 'me'
-        url = '{}/{}'.format(self.base_url, endpoint)
+        endpoint = "me"
+        url = "{}/{}".format(self.base_url, endpoint)
         if self.__user_agent:
-            headers['User-Agent'] = self.__user_agent
-        headers['Accept'] = 'application/json'
+            headers["User-Agent"] = self.__user_agent
+        headers["Accept"] = "application/json"
         response = self.__session.get(
             url=url,
             headers=headers,
-            auth=HTTPBasicAuth(self.__api_key, ''))
+            auth=HTTPBasicAuth(self.__api_key, ""))
         if response.status_code != 200:
-            LOGGER.error('Error status_code = {}'.format(response.status_code))
+            LOGGER.error("Error status_code = {}".format(response.status_code))
             return False
         return True
 
-    @backoff.on_exception(backoff.expo,
-                          (Server5xxError, ConnectionError, MailshakeAPILimitReachedError),
-                          factor=3)
+    @backoff.on_exception(
+        backoff.expo,
+        (Server5xxError, ConnectionError, ChunkedEncodingError, http.client.IncompleteRead,
+         MailshakeAPILimitReachedError, Timeout),
+        factor=3,
+        max_tries=5,
+        on_backoff=lambda details: LOGGER.warning(
+            f"Retrying {details['target'].__name__} after: {details['exception']}")
+    )
     @utils.ratelimit(1, 3)
     def request(self, method, path=None, url=None, json=None, **kwargs):
         """Perform HTTP request"""
@@ -185,50 +198,55 @@ def request(self, method, path=None, url=None, json=None, **kwargs):
         #     self.__verified = self.check_access()
 
         if not url and path:
-            url = '{}/{}'.format(self.base_url, path)
+            url = "{}/{}".format(self.base_url, path)
 
-        if 'endpoint' in kwargs:
-            endpoint = kwargs['endpoint']
-            del kwargs['endpoint']
+        if "endpoint" in kwargs:
+            endpoint = kwargs["endpoint"]
+            del kwargs["endpoint"]
         else:
             endpoint = None
 
-        if 'headers' not in kwargs:
-            kwargs['headers'] = {}
+        if "headers" not in kwargs:
+            kwargs["headers"] = {}
 
-        kwargs['headers']['Accept'] = 'application/json'
+        kwargs["headers"]["Accept"] = "application/json"
 
         if self.__user_agent:
-            kwargs['headers']['User-Agent'] = self.__user_agent
+            kwargs["headers"]["User-Agent"] = self.__user_agent
 
-        if method == 'POST':
-            kwargs['headers']['Content-Type'] = 'application/json'
+        if method == "POST":
+            kwargs["headers"]["Content-Type"] = "application/json"
 
         with metrics.http_request_timer(endpoint) as timer:
+            request_timeout = kwargs.pop("request_timeout", self.request_timeout)
             response = self.__session.request(
                 method=method,
                 url=url,
                 json=json,
-                auth=HTTPBasicAuth(self.__api_key, ''),
+                auth=HTTPBasicAuth(self.__api_key, ""),
+                timeout=request_timeout,
                 **kwargs)
             timer.tags[metrics.Tag.http_status_code] = response.status_code
 
         if response.status_code >= 500:
             raise Server5xxError()
 
+        if response.status_code == 429:
+            raise MailshakeAPILimitReachedError("Rate limit exceeded")
+
         if response.status_code != 200:
             raise_for_error(response)
 
         # pagination details (nextToken) are returned in the body
         next_token = None
         response_body = response.json()
-        if response_body.get('nextToken') != "":
-            next_token = response_body.get('nextToken')
+        if response_body.get("nextToken") != "":
+            next_token = response_body.get("nextToken")
 
         return response_body, next_token
 
     def get(self, path, **kwargs):
-        return self.request('GET', path=path, **kwargs)
+        return self.request("GET", path=path, **kwargs)
 
     def post(self, path, **kwargs):
-        return self.request('POST', path=path, **kwargs)
+        return self.request("POST", path=path, **kwargs)
diff --git a/tap_mailshake/sync.py b/tap_mailshake/sync.py
@@ -1,5 +1,5 @@
 import json
-
+from tap_mailshake.client import REQUEST_TIMEOUT
 import singer
 from singer import (UNIX_SECONDS_INTEGER_DATETIME_PARSING, Transformer,
                     metadata, metrics, utils)
@@ -8,7 +8,6 @@
 
 LOGGER = singer.get_logger()
 
-
 def write_schema(catalog, stream_name):
     stream = catalog.get_stream(stream_name)
     schema = stream.schema.to_dict()
@@ -124,8 +123,12 @@ def sync_endpoint(client,  # pylint: disable=too-many-branches,too-many-nested-b
                   id_fields=None,
                   selected_streams=None,
                   parent=None,
-                  parent_id=None):
+                  parent_id=None,
+                  request_timeout= REQUEST_TIMEOUT):
     # Get the latest bookmark for the stream and set the last_integer/datetime
+    if not start_date:
+        LOGGER.warning(f"No start_date provided for stream '{stream_name}', using default.")
+        start_date = "2019-01-01T00:00:00Z"
     last_datetime = None
     last_integer = None
     data_key = endpoint_config.get('data_key')
@@ -181,7 +184,9 @@ def sync_endpoint(client,  # pylint: disable=too-many-branches,too-many-nested-b
             url=url,
             path=path,
             params=querystring,
-            endpoint=stream_name)
+            endpoint=stream_name,
+            request_timeout=request_timeout,
+        )
 
         # time_extracted: datetime when the data was extracted from the API
         time_extracted = utils.now()
@@ -227,14 +232,14 @@ def sync_endpoint(client,  # pylint: disable=too-many-branches,too-many-nested-b
                     for record in transformed_data:
                         i = 0
                         # Set parent_id
-                        for id_field in id_fields:
-                            parent_id_field = None
-                            if i == 0:
-                                parent_id_field = id_field
-                            if id_field == 'id':
-                                parent_id_field = id_field
-                            i = i + 1
-                        parent_id = record.get(parent_id_field, None)
+                        # Initialize parent_id_field to None in case id_fields is missing or empty
+                        parent_id_field = None
+                        if id_fields:
+                            parent_id_field = 'id' if 'id' in id_fields else id_fields[0]
+                        else:
+                            LOGGER.warning(f"No id_fields provided for stream '{stream_name}'")
+                        # Safely extract parent_id only if parent_id_field is set
+                        parent_id = record.get(parent_id_field) if parent_id_field else None
 
                         # sync_endpoint for child
                         LOGGER.info(
@@ -322,7 +327,9 @@ def get_selected_fields(catalog, stream_name):
 
 
 def sync(client, config, catalog, state):
-    start_date = config.get('start_date', None)
+    start_date = "2019-01-01T00:00:00Z"
+    if 'start_date' in config:
+        start_date = config['start_date']
 
     # Get selected_streams from catalog, based on state last_stream
     #   last_stream = Previous currently synced stream, if the load was interrupted
diff --git a/tap_mailshake/tests/denest_recursive_nodes_data.py b/tap_mailshake/tests/denest_recursive_nodes_data.py
diff --git a/tap_mailshake/tests/test_denest.py b/tap_mailshake/tests/test_denest.py
diff --git a/tap_mailshake/tests/unittests/test_client.py b/tap_mailshake/tests/unittests/test_client.py
