feat(source-amazon-seller-partner): rate limits for fatal status (#42052)

Author: darynaishchenko

airbyte-integrations/connectors/source-amazon-seller-partner/acceptance-test-config.yml

@@ -99,6 +99,10 @@ acceptance_tests:
             bypass_reason: "Data cannot be seeded in the test account, integration tests added for the stream instead"
           - name: GET_VENDOR_FORECASTING_RETAIL_REPORT
             bypass_reason: "Data cannot be seeded in the test account, integration tests added for the stream instead"
+          - name: GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL
+            bypass_reason: "Data for the stream only available for 2 years. Expired for now. Seeding in progress."
+          - name: GET_XML_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL
+            bypass_reason: "Data for the stream only available for 2 years. Expired for now. Seeding in progress."
   incremental:
     tests:
       - config_path: "secrets/config.json"

airbyte-integrations/connectors/source-amazon-seller-partner/integration_tests/expected_records.jsonl

@@ -205,6 +205,13 @@
             }
           }
         }
+      },
+      "wait_to_avoid_fatal_errors": {
+        "title": "Wait between requests to avoid fatal statuses in reports",
+        "type": "boolean",
+        "description": "For report based streams with known amount of requests per time period, this option will use waiting time between requests to avoid fatal statuses in reports. See <a href=\"https://docs.airbyte.com/integrations/sources/amazon-seller-partner#limitations--troubleshooting\" target=\"_blank\">Troubleshooting</a> section for more details",
+        "default": false,
+        "order": 11
       }
     }
   },

airbyte-integrations/connectors/source-amazon-seller-partner/integration_tests/spec_oss.json

@@ -15,7 +15,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: e55879a8-0ef8-4557-abcf-ab34c53ec460
-  dockerImageTag: 4.3.11
+  dockerImageTag: 4.4.0
   dockerRepository: airbyte/source-amazon-seller-partner
   documentationUrl: https://docs.airbyte.com/integrations/sources/amazon-seller-partner
   githubIssueLabel: source-amazon-seller-partner

airbyte-integrations/connectors/source-amazon-seller-partner/metadata.yaml

@@ -3,7 +3,7 @@ requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "4.3.11"
+version = "4.4.0"
 name = "source-amazon-seller-partner"
 description = "Source implementation for Amazon Seller Partner."
 authors = ["Airbyte <[email protected]>"]

airbyte-integrations/connectors/source-amazon-seller-partner/pyproject.toml

@@ -224,7 +224,12 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             if not report_kwargs:
                 report_kwargs.append((stream.report_name, {}))
             for name, options in report_kwargs:
-                kwargs = {"stream_name": name, "report_options": options, **stream_kwargs}
+                kwargs = {
+                    "stream_name": name,
+                    "report_options": options,
+                    "wait_to_avoid_fatal_errors": config.get("wait_to_avoid_fatal_errors", False),
+                    **stream_kwargs,
+                }
                 streams.append(stream(**kwargs))
         return streams
 

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/source.py

@@ -197,6 +197,13 @@
             }
           }
         }
+      },
+      "wait_to_avoid_fatal_errors": {
+        "title": "Wait between requests to avoid fatal statuses in reports",
+        "type": "boolean",
+        "description": "For report based streams with known amount of requests per time period, this option will use waiting time between requests to avoid fatal statuses in reports. See <a href=\"https://docs.airbyte.com/integrations/sources/amazon-seller-partner#limitations--troubleshooting\" target=\"_blank\">Troubleshooting</a> section for more details",
+        "default": false,
+        "order": 11
       }
     }
   },

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/spec.json

@@ -27,6 +27,7 @@
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 from airbyte_protocol.models import FailureType
+from source_amazon_seller_partner.utils import STREAM_THRESHOLD_PERIOD, threshold_period_decorator
 
 REPORTS_API_VERSION = "2021-06-30"
 ORDERS_API_VERSION = "v0"
@@ -222,6 +223,7 @@ def __init__(
         period_in_days: Optional[int],
         replication_end_date: Optional[str],
         report_options: Optional[List[Mapping[str, Any]]] = None,
+        wait_to_avoid_fatal_errors: Optional[bool] = False,
         *args,
         **kwargs,
     ):
@@ -235,6 +237,8 @@ def __init__(
         self._http_method = "GET"
         self._stream_name = stream_name
 
+        self.wait_to_avoid_fatal_errors = wait_to_avoid_fatal_errors
+
     @property
     def name(self):
         return self._stream_name
@@ -382,6 +386,7 @@ def stream_slices(
             }
             start_date = end_date_slice
 
+    @threshold_period_decorator
     def read_records(
         self,
         sync_mode: SyncMode,
@@ -480,6 +485,7 @@ def read_records(
                     f" for period {stream_slice['dataStartTime']}-{stream_slice['dataEndTime']}. "
                     f"This will be read during the next sync. Report ID: {report_id}."
                     f" Error: {error_response}"
+                    " Visit https://docs.airbyte.com/integrations/sources/amazon-seller-partner#limitations--troubleshooting for more info."
                 )
             raise AirbyteTracedException(internal_message=exception_message)
         elif processing_status == ReportProcessingStatus.CANCELLED:
@@ -582,6 +588,10 @@ class FbaStorageFeesReports(IncrementalReportsAmazonSPStream):
 class FulfilledShipmentsReports(IncrementalReportsAmazonSPStream):
     """
     Field definitions: https://sellercentral.amazon.com/gp/help/help.html?itemID=200453120
+
+    Threshold 12
+    Period (minutes) 480
+
     """
 
     report_name = "GET_AMAZON_FULFILLED_SHIPMENTS_DATA_GENERAL"
@@ -681,6 +691,13 @@ def parse_document(self, document):
 
 
 class FbaEstimatedFbaFeesTxtReport(IncrementalReportsAmazonSPStream):
+    """
+
+    Threshold 2000
+    Period (minutes) 60
+
+    """
+
     report_name = "GET_FBA_ESTIMATED_FBA_FEES_TXT_DATA"
 
 
@@ -1110,6 +1127,10 @@ class FbaAfnInventoryReports(IncrementalReportsAmazonSPStream):
     Field definitions: https://developer-docs.amazon.com/sp-api/docs/report-type-values#inventory-reports
     Report has a long-running issue (fails when requested frequently):
     https://github.com/amzn/selling-partner-api-docs/issues/2231
+
+    Threshold 2
+    Period (minutes) 25
+
     """
 
     report_name = "GET_AFN_INVENTORY_DATA"

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/streams.py

@@ -1,13 +1,60 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
+import logging
+import time
 
 from airbyte_cdk.utils import AirbyteTracedException
 from airbyte_protocol.models import FailureType
 
+LOG_LEVEL = logging.getLevelName("INFO")
+LOGGER = logging.getLogger("airbyte")
+
 
 class AmazonConfigException(AirbyteTracedException):
     def __init__(self, **kwargs):
         failure_type: FailureType = FailureType.config_error
         super(AmazonConfigException, self).__init__(failure_type=failure_type, **kwargs)
+
+
+class ReportRateLimits:
+    def __init__(self, threshold: int, period_in_minutes: int):
+        self.threshold = threshold
+        self.period_in_minutes = period_in_minutes
+
+    @property
+    def wait_time_in_seconds(self):
+        return (self.period_in_minutes / self.threshold) * 60
+
+
+# https://github.com/airbytehq/alpha-beta-issues/issues/3717#issuecomment-2203717834
+STREAM_THRESHOLD_PERIOD = {
+    # Threshold sleep logic for GET_AMAZON_FULFILLED_SHIPMENTS_DATA_GENERAL takes 40m between requests,
+    # which is too long for certified connectors. Keeping it as documentation for now.
+    # "GET_AMAZON_FULFILLED_SHIPMENTS_DATA_GENERAL": ReportRateLimits(threshold=12, period_in_minutes=480),
+    "GET_AFN_INVENTORY_DATA": ReportRateLimits(threshold=2, period_in_minutes=25),
+    "GET_FBA_ESTIMATED_FBA_FEES_TXT_DATA": ReportRateLimits(threshold=2000, period_in_minutes=60),
+}
+
+
+def threshold_period_decorator(func):
+    def wrapped(*args, **kwargs):
+        stream_instance = args[0]
+        stream_rate_limits = STREAM_THRESHOLD_PERIOD.get(stream_instance.name)
+
+        # Enable sleeping if stream has known threshold and period or reading without sleeping
+        if stream_instance.wait_to_avoid_fatal_errors and stream_rate_limits:
+            LOGGER.log(
+                LOG_LEVEL,
+                f"Stream {stream_instance.name} has a known rate limits values, applying to avoid rate limits.",
+            )
+
+            for record in func(*args, **kwargs):
+                yield record
+
+                LOGGER.log(LOG_LEVEL, f"Sleeping {stream_rate_limits.wait_time_in_seconds} seconds due to rate limits.")
+                time.sleep(stream_rate_limits.wait_time_in_seconds)
+        else:
+            yield from func(*args, **kwargs)
+
+    return wrapped

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/utils.py

@@ -139,7 +139,8 @@ def test_read_records_retrieve_fatal(self, report_init_kwargs, mocker, requests_
             )
         assert e.value.internal_message == (
             f"Failed to retrieve the report 'GET_TEST_REPORT' for period {stream_start}-{stream_end}. "
-            "This will be read during the next sync. Report ID: some_report_id. Error: Failed to retrieve the report result document."
+            "This will be read during the next sync. Report ID: some_report_id. Error: Failed to retrieve the report result document. "
+            "Visit https://docs.airbyte.com/integrations/sources/amazon-seller-partner#limitations--troubleshooting for more info."
         )
 
     def test_read_records_retrieve_cancelled(self, report_init_kwargs, mocker, requests_mock, caplog):

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/test_streams.py

@@ -61,7 +61,8 @@ To pass the check for Seller and Vendor accounts, you must have access to the [O
 7. For `Start Date`, enter the date in `YYYY-MM-DD` format. The data added on and after this date will be replicated. This field is optional - if not provided or older than 2 years ago from today, the date 2 years ago from today will be used.
 8. For `End Date`, enter the date in `YYYY-MM-DD` format. Any data after this date will not be replicated. This field is optional - if not provided, today's date will be used.
 9. You can specify report options for each stream using **Report Options** section. Available options can be found in corresponding category [here](https://developer-docs.amazon.com/sp-api/docs/report-type-values).
-10. Click `Set up source`.
+10. For `Wait between requests to avoid fatal statuses in reports`, enable if you want to use wating time between requests to avoid fatal statuses in report based streams.
+11. Click `Set up source`.
 
 <!-- /env:cloud -->
 
@@ -76,7 +77,8 @@ To pass the check for Seller and Vendor accounts, you must have access to the [O
 5. For Start Date, enter the date in YYYY-MM-DD format. The data added on and after this date will be replicated. This field is optional - if not provided, the date 2 years ago from today will be used.
 6. For End Date, enter the date in YYYY-MM-DD format. Any data after this date will not be replicated. This field is optional - if not provided, today's date will be used.
 7. You can specify report options for each stream using **Report Options** section. Available options can be found in corresponding category [here](https://developer-docs.amazon.com/sp-api/docs/report-type-values).
-8. Click `Set up source`.
+8. For `Wait between requests to avoid fatal statuses in reports`, enable if you want to use wating time between requests to avoid fatal statuses in report based streams.
+9. Click `Set up source`.
 
 <!-- /env:oss -->
 
@@ -171,24 +173,63 @@ Information about rate limits you may find [here](https://developer-docs.amazon.
 | `array`                  | `array`      |
 | `object`                 | `object`     |
 
+## Limitations & Troubleshooting
+
+### Failed to retrieve the report
+
+```
+Failed to retrieve the report 'YOUR_REPORT_NAME' for period 2024-01-01T12:01:15Z-2024-01-15T12:01:14Z. 
+This will be read during the next sync. Report ID: YOUR_REPORT_ID. Error: Failed to retrieve the report result document.
+```
+
+Requesting reports via Amazon Seller Partner API can lead to failed syncs with error above "Failed to retrieve the report...".
+
+One of the reasons why users face this issue is that report requests were made too often. 
+
+**Solution 1:**
+
+To overcome it you can force use sleeping between requests to avoid fatal statuses while requesting reports.
+
+Steps:
+1. Go to the Set Up page of the connector.
+2. Open optional section.
+3. Enable `Wait between requests to avoid fatal statuses in reports` toggle.
+
+Disadvantages of this approach is that syncs with waiting between requests are much slower than without it. So it is better to create a separate connection only for stream that usually fails with "Failed to retrieve the report..." error. This will help you to avoid affecting streams that worked as expected.
+
+:::note
+
+For now the waiting logic only work for the following streams:
+- GET_AMAZON_FULFILLED_SHIPMENTS_DATA_GENERAL
+- GET_AFN_INVENTORY_DATA
+- GET_FBA_ESTIMATED_FBA_FEES_TXT_DATA
+
+:::
+
+**Solution 2:**
+
+Create a separate connection for streams which usually fail with error above "Failed to retrieve the report..." and disable sync of these streams in the first connection with streams which don't fail because of the error. Adjust the sync time of these two connection to do not overlap. It's recommended to have a time break between syncs in the connections.
+
+
 ## Changelog
 
 <details>
   <summary>Expand to review</summary>
 
 | Version | Date       | Pull Request                                              | Subject                                                                                                                                                                             |
 |:--------|:-----------|:----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| 4.3.11 | 2024-07-13 | [41873](https://github.com/airbytehq/airbyte/pull/41873) | Update dependencies |
-| 4.3.10 | 2024-07-10 | [41345](https://github.com/airbytehq/airbyte/pull/41345) | Update dependencies |
-| 4.3.9 | 2024-07-09 | [41158](https://github.com/airbytehq/airbyte/pull/41158) | Update dependencies |
-| 4.3.8 | 2024-07-08 | [40751](https://github.com/airbytehq/airbyte/pull/40751) | Improve error messaging and turning on alerting |
-| 4.3.7 | 2024-07-06 | [40990](https://github.com/airbytehq/airbyte/pull/40990) | Update dependencies |
-| 4.3.6 | 2024-07-01 | [40590](https://github.com/airbytehq/airbyte/pull/40590) | Add log message when data only accessible to seller accounts, add report id in log message for fatal report status, add check for start date. |
-| 4.3.5 | 2024-06-27 | [40215](https://github.com/airbytehq/airbyte/pull/40215) | Replaced deprecated AirbyteLogger with logging.Logger |
-| 4.3.4 | 2024-06-25 | [40384](https://github.com/airbytehq/airbyte/pull/40384) | Update dependencies |
-| 4.3.3 | 2024-06-22 | [40008](https://github.com/airbytehq/airbyte/pull/40008) | Update dependencies |
-| 4.3.2 | 2024-06-13 | [39441](https://github.com/airbytehq/airbyte/pull/39441) | Update state handling for incremental streams |
-| 4.3.1 | 2024-06-04 | [38969](https://github.com/airbytehq/airbyte/pull/38969) | [autopull] Upgrade base image to v1.2.1 |
+| 4.4.0   | 2024-07-17 | [42052](https://github.com/airbytehq/airbyte/pull/42052)  | Add waiting between requests logic to avoid failed report requests                                                                                                                  |
+| 4.3.11  | 2024-07-13 | [41873](https://github.com/airbytehq/airbyte/pull/41873)  | Update dependencies                                                                                                                                                                 |
+| 4.3.10  | 2024-07-10 | [41345](https://github.com/airbytehq/airbyte/pull/41345)  | Update dependencies                                                                                                                                                                 |
+| 4.3.9   | 2024-07-09 | [41158](https://github.com/airbytehq/airbyte/pull/41158)  | Update dependencies                                                                                                                                                                 |
+| 4.3.8   | 2024-07-08 | [40751](https://github.com/airbytehq/airbyte/pull/40751)  | Improve error messaging and turning on alerting                                                                                                                                     |
+| 4.3.7   | 2024-07-06 | [40990](https://github.com/airbytehq/airbyte/pull/40990)  | Update dependencies                                                                                                                                                                 |
+| 4.3.6   | 2024-07-01 | [40590](https://github.com/airbytehq/airbyte/pull/40590)  | Add log message when data only accessible to seller accounts, add report id in log message for fatal report status, add check for start date.                                       |
+| 4.3.5   | 2024-06-27 | [40215](https://github.com/airbytehq/airbyte/pull/40215)  | Replaced deprecated AirbyteLogger with logging.Logger                                                                                                                               |
+| 4.3.4   | 2024-06-25 | [40384](https://github.com/airbytehq/airbyte/pull/40384)  | Update dependencies                                                                                                                                                                 |
+| 4.3.3   | 2024-06-22 | [40008](https://github.com/airbytehq/airbyte/pull/40008)  | Update dependencies                                                                                                                                                                 |
+| 4.3.2   | 2024-06-13 | [39441](https://github.com/airbytehq/airbyte/pull/39441)  | Update state handling for incremental streams                                                                                                                                       |
+| 4.3.1   | 2024-06-04 | [38969](https://github.com/airbytehq/airbyte/pull/38969)  | [autopull] Upgrade base image to v1.2.1                                                                                                                                             |
 | 4.3.0   | 2024-05-24 | [#00000](https://github.com/airbytehq/airbyte/pull/00000) | Extend the report_options spec config with a `stream_name` attribute                                                                                                                |
 | 4.2.4   | 2024-05-15 | [#38210](https://github.com/airbytehq/airbyte/pull/38210) | Fix `GET_VENDOR_TRAFFIC_REPORT` stream with report option `reportPeriod=DAY`                                                                                                        |
 | 4.2.3   | 2024-05-09 | [#38078](https://github.com/airbytehq/airbyte/pull/38078) | Hide OSS-only streams in report options config for cloud users                                                                                                                      |