🎉 Source Amazon Seller Partner: add replication end date to config (#13059)

Author: ganpatagarwal

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

@@ -25,7 +25,7 @@
 - name: Amazon Seller Partner
   sourceDefinitionId: e55879a8-0ef8-4557-abcf-ab34c53ec460
   dockerRepository: airbyte/source-amazon-seller-partner
-  dockerImageTag: 0.2.19
+  dockerImageTag: 0.2.20
   sourceType: api
   documentationUrl: https://docs.airbyte.io/integrations/sources/amazon-seller-partner
   icon: amazonsellerpartner.svg

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

@@ -213,7 +213,7 @@
               type: "string"
               path_in_connector_config:
               - "client_secret"
-- dockerImage: "airbyte/source-amazon-seller-partner:0.2.19"
+- dockerImage: "airbyte/source-amazon-seller-partner:0.2.20"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/amazon-seller-partner"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/amazon-seller-partner"
@@ -279,6 +279,14 @@
           examples:
           - "2017-01-25T00:00:00Z"
           type: "string"
+        replication_end_date:
+          title: "End Date"
+          description: "UTC date and time in the format 2017-01-25T00:00:00Z. Any\
+            \ data after this date will not be replicated."
+          pattern: "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$|^$"
+          examples:
+          - "2017-01-25T00:00:00Z"
+          type: "string"
         period_in_days:
           title: "Period In Days"
           description: "Will be used for stream slicing for initial full_refresh sync\

airbyte-integrations/connectors/source-amazon-seller-partner/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.2.19
+LABEL io.airbyte.version=0.2.20
 LABEL io.airbyte.name=airbyte/source-amazon-seller-partner

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

@@ -66,6 +66,13 @@
         "examples": ["2017-01-25T00:00:00Z"],
         "type": "string"
       },
+      "replication_end_date": {
+        "title": "End Date",
+        "description": "UTC date and time in the format 2017-01-25T00:00:00Z. Any data after this date will not be replicated.",
+        "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$|^$",
+        "examples": ["2017-01-25T00:00:00Z"],
+        "type": "string"
+      },
       "period_in_days": {
         "title": "Period In Days",
         "description": "Will be used for stream slicing for initial full_refresh sync when no updated state is present for reports that support sliced incremental sync.",

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

@@ -65,6 +65,7 @@ def _get_stream_kwargs(self, config: AmazonSellerPartnerConfig) -> Mapping[str,
             "period_in_days": config.period_in_days,
             "report_options": config.report_options,
             "max_wait_seconds": config.max_wait_seconds,
+            "replication_end_date": config.replication_end_date,
         }
         return stream_kwargs
 

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

@@ -53,6 +53,15 @@ class Config:
         pattern="^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$",
         examples=["2017-01-25T00:00:00Z"],
     )
+
+    replication_end_date: str = Field(
+        None,
+        description="UTC date and time in the format 2017-01-25T00:00:00Z. Any data after this date will not be replicated.",
+        title="End Date",
+        pattern="^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$|^$",
+        examples=["2017-01-25T00:00:00Z"],
+    )
+
     period_in_days: int = Field(
         30,
         description="Will be used for stream slicing for initial full_refresh sync when no updated state is present for reports that support sliced incremental sync.",

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

@@ -46,13 +46,15 @@ def __init__(
         period_in_days: Optional[int],
         report_options: Optional[str],
         max_wait_seconds: Optional[int],
+        replication_end_date: Optional[str],
         *args,
         **kwargs,
     ):
         super().__init__(*args, **kwargs)
 
         self._url_base = url_base.rstrip("/") + "/"
         self._replication_start_date = replication_start_date
+        self._replication_end_date = replication_end_date
         self.marketplace_id = marketplace_id
         self._session.auth = aws_signature
 
@@ -75,6 +77,11 @@ class IncrementalAmazonSPStream(AmazonSPStream, ABC):
     def replication_start_date_field(self) -> str:
         pass
 
+    @property
+    @abstractmethod
+    def replication_end_date_field(self) -> str:
+        pass
+
     @property
     @abstractmethod
     def next_page_token_field(self) -> str:
@@ -97,9 +104,14 @@ def request_params(
             return dict(next_page_token)
 
         params = {self.replication_start_date_field: self._replication_start_date, self.page_size_field: self.page_size}
+
         if self._replication_start_date and self.cursor_field:
             start_date = max(stream_state.get(self.cursor_field, self._replication_start_date), self._replication_start_date)
             params.update({self.replication_start_date_field: start_date})
+
+        if self._replication_end_date:
+            params[self.replication_end_date_field] = self._replication_end_date
+
         return params
 
     def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
@@ -154,13 +166,15 @@ def __init__(
         period_in_days: Optional[int],
         report_options: Optional[str],
         max_wait_seconds: Optional[int],
+        replication_end_date: Optional[str],
         authenticator: HttpAuthenticator = None,
     ):
         self._authenticator = authenticator
         self._session = requests.Session()
         self._url_base = url_base.rstrip("/") + "/"
         self._session.auth = aws_signature
         self._replication_start_date = replication_start_date
+        self._replication_end_date = replication_end_date
         self.marketplace_id = marketplace_id
         self.period_in_days = period_in_days
         self._report_options = report_options
@@ -223,12 +237,21 @@ def _report_data(
     ) -> Mapping[str, Any]:
         replication_start_date = max(pendulum.parse(self._replication_start_date), pendulum.now("utc").subtract(days=90))
 
-        return {
+        params = {
             "reportType": self.name,
             "marketplaceIds": [self.marketplace_id],
             "dataStartTime": replication_start_date.strftime(DATE_TIME_FORMAT),
         }
 
+        if self._replication_end_date and sync_mode == SyncMode.full_refresh:
+            params["dataEndTime"] = self._replication_end_date
+            # if replication_start_date is older than 90 days(from current date), we are overriding the value above.
+            # when replication_end_date is present, we should use the user provided replication_start_date.
+            # user may provide a date range which is older than 90 days.
+            params["dataStartTime"] = self._replication_start_date
+
+        return params
+
     def _create_report(
         self,
         sync_mode: SyncMode,
@@ -539,6 +562,8 @@ def stream_slices(
 
         start_date = pendulum.parse(self._replication_start_date)
         end_date = pendulum.now()
+        if self._replication_end_date and sync_mode == SyncMode.full_refresh:
+            end_date = pendulum.parse(self._replication_end_date)
 
         if stream_state:
             state = stream_state.get(self.cursor_field)
@@ -648,6 +673,7 @@ class Orders(IncrementalAmazonSPStream):
     primary_key = "AmazonOrderId"
     cursor_field = "LastUpdateDate"
     replication_start_date_field = "LastUpdatedAfter"
+    replication_end_date_field = "LastUpdatedBefore"
     next_page_token_field = "NextToken"
     page_size_field = "MaxResultsPerPage"
     default_backoff_time = 60
@@ -686,16 +712,11 @@ class VendorDirectFulfillmentShipping(AmazonSPStream):
     name = "VendorDirectFulfillmentShipping"
     primary_key = None
     replication_start_date_field = "createdAfter"
+    replication_end_date_field = "createdBefore"
     next_page_token_field = "nextToken"
     page_size_field = "limit"
     time_format = "%Y-%m-%dT%H:%M:%SZ"
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.replication_start_date_field = max(
-            pendulum.parse(self._replication_start_date), pendulum.now("utc").subtract(days=7, hours=1)
-        ).strftime(self.time_format)
-
     def path(self, **kwargs) -> str:
         return f"vendor/directFulfillment/shipping/{VENDORS_API_VERSION}/shippingLabels"
 
@@ -704,7 +725,15 @@ def request_params(
     ) -> MutableMapping[str, Any]:
         params = super().request_params(stream_state=stream_state, next_page_token=next_page_token, **kwargs)
         if not next_page_token:
-            params.update({"createdBefore": pendulum.now("utc").strftime(self.time_format)})
+            end_date = pendulum.now("utc").strftime(self.time_format)
+            if self._replication_end_date:
+                end_date = self._replication_end_date
+
+            start_date = max(pendulum.parse(self._replication_start_date), pendulum.parse(end_date).subtract(days=7, hours=1)).strftime(
+                self.time_format
+            )
+
+            params.update({self.replication_start_date_field: start_date, self.replication_end_date_field: end_date})
         return params
 
     def parse_response(self, response: requests.Response, stream_state: Mapping[str, Any], **kwargs) -> Iterable[Mapping]:

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

@@ -22,6 +22,7 @@ def reports_stream():
         url_base="https://test.url",
         aws_signature=aws_signature,
         replication_start_date="2017-01-25T00:00:00Z",
+        replication_end_date="2017-02-25T00:00:00Z",
         marketplace_id="id",
         authenticator=None,
         period_in_days=0,

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

@@ -21,6 +21,7 @@ def connector_source():
 def connector_config():
     return AmazonSellerPartnerConfig(
         replication_start_date="2017-01-25T00:00:00Z",
+        replication_end_date="2017-02-25T00:00:00Z",
         refresh_token="Atzr|IwEBIP-abc123",
         lwa_app_id="amzn1.application-oa2-client.abc123",
         lwa_client_secret="abc123",

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

@@ -19,6 +19,7 @@ def reports_stream(marketplace_id):
         url_base="https://test.url",
         aws_signature=aws_signature,
         replication_start_date="2010-01-25T00:00:00Z",
+        replication_end_date="2017-02-25T00:00:00Z",
         marketplace_id=marketplace_id,
         authenticator=None,
         period_in_days=0,

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

@@ -90,6 +90,7 @@ This source is capable of syncing the following tables and their data:
 
 | Version  | Date       | Pull Request                                               | Subject                                                                |
 |:---------|:-----------|:-----------------------------------------------------------|:-----------------------------------------------------------------------|
+| `0.2.20` | 2022-05-30 | [\#13059](https://github.com/airbytehq/airbyte/pull/13059) | Add replication end date to config                                     |
 | `0.2.19` | 2022-05-24 | [\#13119](https://github.com/airbytehq/airbyte/pull/13119) | Add OAuth2.0 support                                                   |
 | `0.2.18` | 2022-05-06 | [\#12663](https://github.com/airbytehq/airbyte/pull/12663) | Add GET_XML_BROWSE_TREE_DATA report                                    |
 | `0.2.17` | 2022-05-19 | [\#12946](https://github.com/airbytehq/airbyte/pull/12946) | Add throttling exception managing in Orders streams                    |