Amazon Ads: changes for increasing timeout for long taking report generation

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -15,7 +15,7 @@
 - name: Amazon Ads
   sourceDefinitionId: c6b0a29e-1da9-4512-9002-7bfd0cba2246
   dockerRepository: airbyte/source-amazon-ads
-  dockerImageTag: 0.1.3
+  dockerImageTag: 0.1.4
   documentationUrl: https://docs.airbyte.io/integrations/sources/amazon-ads
   icon: amazonads.svg
   sourceType: api

airbyte-integrations/connectors/source-amazon-ads/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.3
-LABEL io.airbyte.name=airbyte/source-amazon-ads
+LABEL io.airbyte.version=0.1.4
+LABEL io.airbyte.name=airbyte/source-amazon-ads

airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/spec.py

@@ -65,6 +65,18 @@ class Config:
         description="profile Ids you want to fetch data for",
     )
 
+    report_wait_timeout: int = Field(
+        None,
+        name="Report Wait Timeout",
+        description="Timeout duration for Reports"
+    )
+
+    report_generation_max_retries: int = Field(
+        None,
+        name="Report Geration Maximum Retries",
+        description="Maximum retries Airbyte will attempt for fetching Report Data",
+    )
+
     @classmethod
     def schema(cls, **kvargs):
         schema = super().schema(**kvargs)

airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/streams/report_streams/report_streams.py

@@ -1,7 +1,7 @@
 #
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
-
+ 
 import json
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
@@ -102,11 +102,18 @@ class ReportStream(BasicAmazonAdsStream, ABC):
     # Format used to specify metric generation date over Amazon Ads API.
     REPORT_DATE_FORMAT = "%Y%m%d"
     cursor_field = "reportDate"
+    #Maximum number of retries that Airbyte attempt to fetch data.
+    REPORT_GENERATION_MAX_RETRIES = 5
 
     def __init__(self, config: AmazonAdsConfig, profiles: List[Profile], authenticator: Oauth2Authenticator):
         self._authenticator = authenticator
         self._session = requests.Session()
         self._model = self._generate_model()
+        if(config and config.report_wait_timeout and config.report_wait_timeout > 30):
+            self.REPORT_WAIT_TIMEOUT = timedelta(minutes=config.report_wait_timeout).total_seconds
+        if(config and config.report_generation_max_retries and config.report_generation_max_retries > 5):
+            self.REPORT_GENERATION_MAX_RETRIES = config.report_generation_max_retries
+
         # Set start date from config file, should be in UTC timezone.
         self._start_date = pendulum.parse(config.start_date).set(tz="UTC") if config.start_date else None
         super().__init__(config, profiles)
@@ -150,25 +157,34 @@ def read_records(
                     metric=metric_object,
                 ).dict()
 
-    @backoff.on_exception(
-        backoff.expo,
-        ReportGenerationFailure,
-        max_tries=5,
-    )
+    def backoff_max_time(func):
+        def wrapped(self, *args, **kwargs):
+            return backoff.on_exception(
+                backoff.constant,
+                RetryableException,
+                max_time = self.REPORT_WAIT_TIMEOUT
+            )(func)(self, *args, **kwargs)
+        return wrapped
+
+    def backoff_max_tries(func):
+        def wrapped(self, *args, **kwargs):
+            return backoff.on_exception(
+                backoff.expo,
+                ReportGenerationFailure,
+                max_tries = self.REPORT_GENERATION_MAX_RETRIES
+            )(func)(self, *args, **kwargs)
+        return wrapped
+
+    @backoff_max_tries
     def _init_and_try_read_records(self, report_date):
         report_infos = self._init_reports(report_date)
         logger.info(f"Waiting for {len(report_infos)} report(s) to be generated")
         self._try_read_records(report_infos)
         return report_infos
 
-    @backoff.on_exception(
-        backoff.constant,
-        RetryableException,
-        max_time=REPORT_WAIT_TIMEOUT,
-    )
+    @backoff_max_time
     def _try_read_records(self, report_infos):
         incomplete_report_infos = self._incomplete_report_infos(report_infos)
-
         logger.info(f"Checking report status, {len(incomplete_report_infos)} report(s) remaining")
         for report_info in incomplete_report_infos:
             report_status, download_url = self._check_status(report_info)

docs/integrations/sources/amazon-ads.md

@@ -76,6 +76,7 @@ Start date used for generating reports starting from the specified start date. S
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| `0.1.4` | 2022-02-21 | [\#10513](https://github.com/airbytehq/airbyte/pull/10513) | `Increasing REPORT_WAIT_TIMEOUT for supporting report generation which takes longer time ` |
 | `0.1.3` | 2021-12-28 | [\#8388](https://github.com/airbytehq/airbyte/pull/8388) | `Add retry if recoverable error  occured for reporting stream processing` |
 | `0.1.2` | 2021-10-01 | [\#6367](https://github.com/airbytehq/airbyte/pull/6461) | `Add option to pull data for different regions. Add option to choose profiles we want to pull data. Add lookback` |
 | `0.1.1` | 2021-09-22 | [\#6367](https://github.com/airbytehq/airbyte/pull/6367) | `Add seller and vendor filters to profiles stream` |