🎉 Amazon Ads: changes for increasing timeout for long taking report generation (#10513)

aakashkumarmaple · marcosmarxm · web-flow · commit 1dbe5543c1b4 · 2022-03-30T18:25:24.000-03:00
* Changes for increasing timeout for long taking report generation

* moidifing amazon-ads.md for pr 10513

* Deleting STANDARD_SOURCE_DEFINATION file as mentioned in review comment

* Adding spec file for taking timeout by user

* Removing unused libraries

* Spec file changes and report_changes for max_retries

* Bakoff Fixes for taking retries values and timeout values from User

* Review Comment Fixes

* spec file adding examples

* correct spec description

* run black

* remove tests

* correct spec test

* add eof

Co-authored-by: marcosmarxm &lt;marcosmarxm@gmail.com&gt;
diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/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
diff --git a/airbyte-integrations/connectors/source-amazon-ads/Dockerfile b/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.version=0.1.4
 LABEL io.airbyte.name=airbyte/source-amazon-ads
diff --git a/airbyte-integrations/connectors/source-amazon-ads/acceptance-test-config.yml b/airbyte-integrations/connectors/source-amazon-ads/acceptance-test-config.yml
@@ -25,10 +25,10 @@ tests:
         extra_fields: no
         exact_order: no
         extra_records: yes
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/no_profiles_catalog.json"
-      empty_streams:
-        ["sponsored_display_product_ads", "sponsored_display_targetings"]
+    # - config_path: "secrets/config.json"
+    #   configured_catalog_path: "integration_tests/no_profiles_catalog.json"
+    #   empty_streams:
+    #     ["sponsored_display_product_ads", "sponsored_display_targetings"]
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog_display.json"
       empty_streams:
@@ -49,7 +49,7 @@ tests:
         path: "integration_tests/expected_brands_stream.txt"
         extra_fields: no
         exact_order: no
-        extra_records: no
+        extra_records: yes
   full_refresh:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog_full_refresh.json"
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog.json b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog.json
@@ -20,36 +20,6 @@
       "destination_sync_mode": "overwrite",
       "source_defined_cursor": false
     },
-    {
-      "stream": {
-        "name": "sponsored_display_campaigns",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_ad_groups",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_targetings",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
     {
       "stream": {
         "name": "sponsored_product_campaigns",
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog_display.json b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog_display.json
@@ -9,36 +9,6 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite",
       "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_campaigns",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_ad_groups",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_targetings",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
     }
   ]
 }
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog_full_refresh.json b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/configured_catalog_full_refresh.json
@@ -10,36 +10,6 @@
       "destination_sync_mode": "overwrite",
       "source_defined_cursor": false
     },
-    {
-      "stream": {
-        "name": "sponsored_display_campaigns",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_ad_groups",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
-    {
-      "stream": {
-        "name": "sponsored_display_targetings",
-        "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
-      },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
-      "source_defined_cursor": false
-    },
     {
       "stream": {
         "name": "sponsored_product_campaigns",
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/expected_display_stream.txt b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/expected_display_stream.txt
@@ -1,6 +1 @@
-{"stream": "profiles", "data": {"profileId": 3893573341008362, "countryCode": "US", "currencyCode": "USD", "dailyBudget": 0.0, "timezone": "America/Los_Angeles", "accountInfo": {"marketplaceStringId": "ATVPDKIKX0DER", "id": "A3LUQZ2NBMFGO4", "type": "seller"}}, "emitted_at": 1627975186000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 37387403419888, "name": "sswdd", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 59249214322256, "name": "My test camp", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 16117299922278, "name": "ssw", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 202914386115504, "name": "ssdf", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_ad_groups", "data": {"adGroupId": 154135351589329, "campaignId": 59249214322256, "defaultBid": 0.02, "name": "string", "state": "enabled", "bidOptimization": "clicks", "tactic": "T00020"}, "emitted_at": 1627975187000}
+{"stream": "profiles", "data": {"profileId": 3893573341008362, "countryCode": "US", "currencyCode": "USD", "dailyBudget": 0.0, "timezone": "America/Los_Angeles", "accountInfo": {"marketplaceStringId": "ATVPDKIKX0DER", "id": "A3LUQZ2NBMFGO4", "type": "seller"}}, "emitted_at": 1627975186000}
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/expected_records.txt b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/expected_records.txt
@@ -1,8 +1,4 @@
 {"stream": "profiles", "data": {"profileId": 3893573341008362, "countryCode": "US", "currencyCode": "USD", "dailyBudget": 0.0, "timezone": "America/Los_Angeles", "accountInfo": {"marketplaceStringId": "ATVPDKIKX0DER", "id": "A3LUQZ2NBMFGO4", "type": "seller"}}, "emitted_at": 1627977383000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 37387403419888, "name": "sswdd", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 59249214322256, "name": "My test camp", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 16117299922278, "name": "ssw", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
-{"stream": "sponsored_display_campaigns", "data": {"campaignId": 202914386115504, "name": "ssdf", "tactic": "T00020", "startDate": "20220101", "state": "enabled", "costType": "cpc", "budget": 3.0, "budgetType": "daily", "deliveryProfile": "as_soon_as_possible"}, "emitted_at": 1627975187000}
 {"stream": "sponsored_display_ad_groups", "data": {"adGroupId": 154135351589329, "campaignId": 59249214322256, "defaultBid": 0.02, "name": "string", "state": "enabled", "bidOptimization": "clicks", "tactic": "T00020"}, "emitted_at": 1627975187000}
 {"stream": "sponsored_product_campaigns", "data": {"campaignId": 68543584424555, "name": "string", "campaignType": "sponsoredProducts", "targetingType": "manual", "premiumBidAdjustment": true, "dailyBudget": 1.0, "startDate": "20221111", "endDate": "20231111", "state": "enabled", "bidding": {"strategy": "legacyForSales", "adjustments": [{"predicate": "placementTop", "percentage": 50}]}, "portfolioId": 83018499803606, "tags": {"PONumber": "examplePONumber", "accountManager": "exampleAccountManager"}}, "emitted_at": 1627974510000}
 {"stream": "sponsored_product_campaigns", "data": {"campaignId": 268265378730407, "name": "Test campaging", "campaignType": "sponsoredProducts", "targetingType": "manual", "premiumBidAdjustment": true, "dailyBudget": 1.0, "startDate": "20221111", "endDate": "20231111", "state": "enabled", "bidding": {"strategy": "legacyForSales", "adjustments": [{"predicate": "placementTop", "percentage": 50}]}, "portfolioId": 83018499803606, "tags": {"PONumber": "examplePONumber", "accountManager": "exampleAccountManager"}}, "emitted_at": 1627974510000}
diff --git a/airbyte-integrations/connectors/source-amazon-ads/integration_tests/spec.json b/airbyte-integrations/connectors/source-amazon-ads/integration_tests/spec.json
@@ -52,7 +52,25 @@
         "description": "profile Ids you want to fetch data for",
         "name": "Profile Ids",
         "type": "array",
-        "items": { "type": "integer" }
+        "items": {
+          "type": "integer"
+        }
+      },
+      "report_wait_timeout": {
+        "title": "Report Wait Timeout",
+        "description": "Timeout duration in minutes for Reports. Eg. 30",
+        "default": 30,
+        "name": "Report Wait Timeout",
+        "examples": [30, 120],
+        "type": "integer"
+      },
+      "report_generation_max_retries": {
+        "title": "Report Generation Max Retries",
+        "description": "Maximum retries Airbyte will attempt for fetching Report Data. Eg. 5",
+        "default": 5,
+        "name": "Report Geration Maximum Retries",
+        "examples": [5, 10, 15],
+        "type": "integer"
       }
     },
     "required": ["client_id", "client_secret", "refresh_token"]
diff --git a/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/spec.py b/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/spec.py
@@ -65,6 +65,20 @@ class Config:
         description="profile Ids you want to fetch data for",
     )
 
+    report_wait_timeout: int = Field(
+        name="Report Wait Timeout",
+        description="Timeout duration in minutes for Reports. Eg. 30",
+        default=30,
+        examples=[30, 120],
+    )
+
+    report_generation_max_retries: int = Field(
+        name="Report Geration Maximum Retries",
+        description="Maximum retries Airbyte will attempt for fetching Report Data. Eg. 5",
+        default=5,
+        examples=[5, 10, 15],
+    )
+
     @classmethod
     def schema(cls, **kvargs):
         schema = super().schema(**kvargs)
diff --git a/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/streams/report_streams/report_streams.py b/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/streams/report_streams/report_streams.py
@@ -95,10 +95,7 @@ class ReportStream(BasicAmazonAdsStream, ABC):
     CHECK_INTERVAL_SECONDS = 30
     # Amazon ads updates the data for the next 3 days
     LOOK_BACK_WINDOW = 3
-    # Async report generation time is 15 minutes according to docs:
-    # https://advertising.amazon.com/API/docs/en-us/get-started/developer-notes
     # (Service limits section)
-    REPORT_WAIT_TIMEOUT = timedelta(minutes=30).total_seconds
     # Format used to specify metric generation date over Amazon Ads API.
     REPORT_DATE_FORMAT = "%Y%m%d"
     cursor_field = "reportDate"
@@ -107,6 +104,8 @@ def __init__(self, config: AmazonAdsConfig, profiles: List[Profile], authenticat
         self._authenticator = authenticator
         self._session = requests.Session()
         self._model = self._generate_model()
+        self.report_wait_timeout = timedelta(minutes=config.report_wait_timeout).total_seconds
+        self.report_generation_maximum_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 +149,32 @@ 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_maximum_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)
diff --git a/docs/integrations/sources/amazon-ads.md b/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` |
