Source Amazon Seller Partner: add stream FbaReplacementsReports

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

@@ -38,6 +38,8 @@ tests:
 #      configured_catalog_path: "integration_tests/configured_catalog_brand_analytics_repeat_purchase.json.json"
 #    - config_path: "secrets/config.json"
 #      configured_catalog_path: "integration_tests/configured_catalog_brand_analytics_search_terms.json"
+#    - config_path: "secrets/config.json"
+#      configured_catalog_path: "integration_tests/configured_catalog_fba_fulfillment_customer_shipment_replacement_data.json"
 # TODO: uncomment when Orders (or any other incremental) stream is filled with data
 #  incremental:
 #   - config_path: "secrets/config.json"

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

@@ -0,0 +1,66 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "GET_FBA_FULFILLMENT_CUSTOMER_SHIPMENT_REPLACEMENT_DATA",
+        "json_schema": {
+          "title": "FBA Fulfillment Customer Shipment Replacement Data",
+          "description": "FBA Fulfillment Customer Shipment Replacement Data Reports",
+          "type": "object",
+          "$schema": "http://json-schema.org/draft-07/schema#",
+          "properties": {
+            "shipment-date": {
+              "title": "Date",
+              "description": "DD-MON-YYYY. Date of free replacement shipment",
+              "type": ["null", "string"],
+              "format": "date-time"
+            },
+            "sku": {
+              "title": "Merchant SKU",
+              "description": "Seller's item identifier",
+              "type": ["null", "string"]
+            },
+            "asin": {
+              "title": "ASIN",
+              "description": "Amazon Standard Inventory Number",
+              "type": ["null", "string"]
+            },
+            "fulfillment-center-id": {
+              "title": "Warehouse ID",
+              "description": "Fulfillment center shipping this unit.",
+              "type": ["null", "string"]
+            },
+            "original-fulfillment-center-id": {
+              "title": "Original Warehouse ID",
+              "description": "Original fulfillment center shipping free replaced item.",
+              "type": ["null", "string"]
+            },
+            "quantity": {
+              "title": "Quantity",
+              "description": "Units shipped in replacement shipment",
+              "type": ["null", "string"]
+            },
+            "replacement-reason-code": {
+              "title": "Replacement Reason Code",
+              "description": "Reason for replacement",
+              "type": ["null", "string"]
+            },
+            "replacement-customer-order-id": {
+              "title": "Replacement Customer Order Id",
+              "description": "Order ID of the replacement order",
+              "type": ["null", "string"]
+            },
+            "original-amazon-order-id": {
+              "title": "Original Customer Order ID",
+              "description": "Order ID of original shipment",
+              "type": ["null", "string"]
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    }
+  ]
+}

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

@@ -0,0 +1,54 @@
+{
+  "title": "FBA Fulfillment Customer Shipment Replacement Data",
+  "description": "FBA Fulfillment Customer Shipment Replacement Data Reports",
+  "type": "object",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "shipment-date": {
+      "title": "Date",
+      "description": "DD-MON-YYYY. Date of free replacement shipment",
+      "type": ["null", "string"],
+      "format": "date-time"
+    },
+    "sku": {
+      "title": "Merchant SKU",
+      "description": "Seller's item identifier",
+      "type": ["null", "string"]
+    },
+    "asin": {
+      "title": "ASIN",
+      "description": "Amazon Standard Inventory Number",
+      "type": ["null", "string"]
+    },
+    "fulfillment-center-id": {
+      "title": "Warehouse ID",
+      "description": "Fulfillment center shipping this unit.",
+      "type": ["null", "string"]
+    },
+    "original-fulfillment-center-id": {
+      "title": "Original Warehouse ID",
+      "description": "Original fulfillment center shipping free replaced item.",
+      "type": ["null", "string"]
+    },
+    "quantity": {
+      "title": "Quantity",
+      "description": "Units shipped in replacement shipment",
+      "type": ["null", "string"]
+    },
+    "replacement-reason-code": {
+      "title": "Replacement Reason Code",
+      "description": "Reason for replacement",
+      "type": ["null", "string"]
+    },
+    "replacement-customer-order-id": {
+      "title": "Replacement Customer Order Id",
+      "description": "Order ID of the replacement order",
+      "type": ["null", "string"]
+    },
+    "original-amazon-order-id": {
+      "title": "Original Customer Order ID",
+      "description": "Order ID of original shipment",
+      "type": ["null", "string"]
+    }
+  }
+}

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

@@ -22,6 +22,7 @@
     BrandAnalyticsSearchTermsReports,
     FbaInventoryReports,
     FbaOrdersReports,
+    FbaReplacementsReports,
     FbaShipmentsReports,
     FlatFileOpenListingsReports,
     FlatFileOrdersReports,
@@ -147,6 +148,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             FbaInventoryReports(**stream_kwargs),
             FbaOrdersReports(**stream_kwargs),
             FbaShipmentsReports(**stream_kwargs),
+            FbaReplacementsReports(**stream_kwargs),
             FlatFileOpenListingsReports(**stream_kwargs),
             FlatFileOrdersReports(**stream_kwargs),
             FlatFileOrdersReportsByLastUpdate(**stream_kwargs),

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

@@ -319,6 +319,8 @@ def read_records(
             seconds_waited = (pendulum.now("utc") - start_time).seconds
             is_processed = report_payload.get("processingStatus") not in ["IN_QUEUE", "IN_PROGRESS"]
             is_done = report_payload.get("processingStatus") == "DONE"
+            is_cancelled = report_payload.get("processingStatus") == "CANCELLED"
+            is_fatal = report_payload.get("processingStatus") == "FATAL"
             time.sleep(self.sleep_seconds)
 
         if is_done:
@@ -332,8 +334,12 @@ def read_records(
             )
             response = self._send_request(request)
             yield from self.parse_response(response)
+        elif is_fatal:
+            raise Exception(f"The report for stream '{self.name}' was aborted due to a fatal error")
+        elif is_cancelled:
+            logger.warn(f"The report for stream '{self.name}' was cancelled or there is no data to return")
         else:
-            logger.warn(f"There are no report document related in stream `{self.name}`. Report body {report_payload}")
+            raise Exception(f"Unknown response for stream `{self.name}`. Response body {report_payload}")
 
 
 class MerchantListingsReports(ReportsAmazonSPStream):
@@ -384,6 +390,14 @@ class FbaShipmentsReports(ReportsAmazonSPStream):
     name = "GET_FBA_FULFILLMENT_REMOVAL_SHIPMENT_DETAIL_DATA"
 
 
+class FbaReplacementsReports(ReportsAmazonSPStream):
+    """
+    Field definitions: https://sellercentral.amazon.com/help/hub/reference/200453300
+    """
+
+    name = "GET_FBA_FULFILLMENT_CUSTOMER_SHIPMENT_REPLACEMENT_DATA"
+
+
 class VendorInventoryHealthReports(ReportsAmazonSPStream):
     name = "GET_VENDOR_INVENTORY_HEALTH_AND_PLANNING_REPORT"
 
@@ -651,7 +665,7 @@ class VendorDirectFulfillmentShipping(AmazonSPStream):
     """
 
     name = "VendorDirectFulfillmentShipping"
-    primary_key = [["labelData", "packageIdentifier"]]
+    primary_key = None
     replication_start_date_field = "createdAfter"
     next_page_token_field = "nextToken"
     page_size_field = "limit"

docs/integrations/sources/amazon-seller-partner.md

@@ -16,6 +16,7 @@ This source is capable of syncing the following streams:
 - [Order Report (by order date and by last update)](https://sellercentral.amazon.com/gp/help/help.html?itemID=201648780)
 - [All Listings](https://github.com/amzn/selling-partner-api-docs/blob/main/references/reports-api/reporttype-values.md#inventory-reports)
 - [FBA Inventory Reports](https://sellercentral.amazon.com/gp/help/200740930)
+- [FBA Replacements Report](https://sellercentral.amazon.com/help/hub/reference/200453300)
 - [Amazon-Fulfilled Shipments Report](https://sellercentral.amazon.com/gp/help/help.html?itemID=200453120)
 - [Open Listings Report](https://github.com/amzn/selling-partner-api-docs/blob/main/references/reports-api/reporttype-values.md#inventory-reports)
 - [Removal Order Detail Report (overview)](https://sellercentral.amazon.com/gp/help/help.html?itemID=200989110)
@@ -67,6 +68,7 @@ Information about rate limits you may find [here](https://github.com/amzn/sellin
 
 | Version  | Date       | Pull Request                                             | Subject                                                                |
 | :------- | :--------- | :------------------------------------------------------- | :--------------------------------------------------------------------- |
+| `0.2.15` | 2022-01-25 | [\#9789](https://github.com/airbytehq/airbyte/pull/9789) | Add stream FbaReplacementsReports                                      |
 | `0.2.14` | 2022-01-19 | [\#9621](https://github.com/airbytehq/airbyte/pull/9621) | Add GET_FLAT_FILE_ALL_ORDERS_DATA_BY_LAST_UPDATE_GENERAL report        |
 | `0.2.13` | 2022-01-18 | [\#9581](https://github.com/airbytehq/airbyte/pull/9581) | Change createdSince parameter to dataStartTime                         |
 | `0.2.12` | 2022-01-05 | [\#9312](https://github.com/airbytehq/airbyte/pull/9312) | Add all remaining brand analytics report streams                       |