🎉 Source Amazon Ads: Skip API error "Report date is too far in the past."

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.16
+LABEL io.airbyte.version=0.1.17
 LABEL io.airbyte.name=airbyte/source-amazon-ads

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

@@ -15,7 +15,6 @@
 import backoff
 import pendulum
 import requests
-from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams.http.auth import Oauth2Authenticator
 from pendulum import Date
@@ -24,8 +23,6 @@
 from source_amazon_ads.streams.common import BasicAmazonAdsStream
 from source_amazon_ads.utils import iterate_one_by_one
 
-logger = AirbyteLogger()
-
 
 class RecordType(str, Enum):
     CAMPAIGNS = "campaigns"
@@ -80,6 +77,10 @@ class ReportInitFailure(RetryableException):
     pass
 
 
+class ReportInitFatal(Exception):
+    pass
+
+
 class TooManyRequests(Exception):
     """
     Custom exception occured when response with 429 status code received
@@ -176,14 +177,14 @@ def wrapped(self, *args, **kwargs):
     @backoff_max_tries
     def _init_and_try_read_records(self, profile: Profile, report_date):
         report_infos = self._init_reports(profile, report_date)
-        logger.info(f"Waiting for {len(report_infos)} report(s) to be generated")
+        self.logger.info(f"Waiting for {len(report_infos)} report(s) to be generated")
         self._try_read_records(report_infos)
         return report_infos
 
     @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")
+        self.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)
             report_info.status = report_status
@@ -358,16 +359,17 @@ def _init_reports(self, profile: Profile, report_date: str) -> List[ReportInfo]:
             # subtypes have mutualy excluded parameters so we requesting
             # different metric list for each record.
             record_type = record_type.split("_")[0]
-            logger.info(f"Initiating report generation for {profile.profileId} profile with {record_type} type for {report_date} date")
+            self.logger.info(f"Initiating report generation for {profile.profileId} profile with {record_type} type for {report_date} date")
             response = self._send_http_request(
                 urljoin(self._url, self.report_init_endpoint(record_type)),
                 profile.profileId,
                 report_init_body,
             )
             if response.status_code != HTTPStatus.ACCEPTED:
-                raise ReportInitFailure(
-                    f"Unexpected HTTP status code {response.status_code} when registering {record_type}, {type(self).__name__} for {profile.profileId} profile: {response.text}"
-                )
+                error_msg = f"Unexpected HTTP status code {response.status_code} when registering {record_type}, {type(self).__name__} for {profile.profileId} profile: {response.text}"
+                if self._check_report_date_error(response):
+                    raise ReportInitFatal(error_msg)
+                raise ReportInitFailure(error_msg)
 
             response = ReportInitResponse.parse_raw(response.text)
             report_infos.append(
@@ -379,7 +381,7 @@ def _init_reports(self, profile: Profile, report_date: str) -> List[ReportInfo]:
                     metric_objects=[],
                 )
             )
-            logger.info("Initiated successfully")
+            self.logger.info("Initiated successfully")
 
         return report_infos
 
@@ -401,3 +403,21 @@ def get_error_display_message(self, exception: BaseException) -> Optional[str]:
         if isinstance(exception, ReportGenerationInProgress):
             return f'Report(s) generation time took more than {self.report_wait_timeout} minutes, please increase the "report_wait_timeout" parameter in configuration.'
         return super().get_error_display_message(exception)
+
+    def _check_report_date_error(self, response) -> bool:
+        """
+        Check if the connector received an error: 'Report date is too far in the past. Reports are only available for 60 days.'
+
+        In theory, it does not have to get such an error because the connector correctly calculates the start date,
+        but from practice, we can still catch such errors from time to time.
+
+        Anyway connector self recovers in subsequent sync because it will re-evaluate the start date and will succeed.
+        """
+
+        if response.status_code == 406:
+            try:
+                response_json = response.json()
+            except ValueError:
+                return False
+            return response_json.get("details", "").startswith("Report date is too far in the past.")
+        return False

docs/integrations/sources/amazon-ads.md

@@ -90,6 +90,7 @@ Information about expected report generation waiting time you may find [here](ht
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                           |
 |:--------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------|
+| 0.1.17 | 2022-08-24 | [15921](https://github.com/airbytehq/airbyte/pull/15921)    | Don't re-try on "Report date is too far in the past" error                                                        |
 | 0.1.16 | 2022-08-23 | [15822](https://github.com/airbytehq/airbyte/pull/15822)    | Set default value for 'region' if needed                                                                          |
 | 0.1.15 | 2022-08-20 | [15816](https://github.com/airbytehq/airbyte/pull/15816)    | Update STATE of incremental sync if no records                                                                    |
 | 0.1.14 | 2022-08-15 | [15637](https://github.com/airbytehq/airbyte/pull/15637)    | Generate slices by lazy evaluation                                                                                |