Source Cart.com: implement Central API Router access method and improve backoff policy (airbytehq#16612)

* add central_api_route param in spec.json

* add central api router and better backoff policy

* move logic to auth class

* finish central api router

* update auth methods

* fix tests

* readded config tests using access token

* auto-bump connector version [ci skip]

Co-authored-by: Octavia Squidington III <[email protected]>

Author: 2 people

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

@@ -136,7 +136,7 @@
 - name: Cart.com
   sourceDefinitionId: bb1a6d31-6879-4819-a2bd-3eed299ea8e2
   dockerRepository: airbyte/source-cart
-  dockerImageTag: 0.1.6
+  dockerImageTag: 0.2.0
   documentationUrl: https://docs.airbyte.io/integrations/sources/cart
   icon: cart.svg
   sourceType: api

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

@@ -1473,30 +1473,80 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-cart:0.1.6"
+- dockerImage: "airbyte/source-cart:0.2.0"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/cart"
     connectionSpecification:
       $schema: "http://json-schema.org/draft-07/schema#"
       title: "Cart.com Spec"
       type: "object"
       required:
-      - "access_token"
       - "start_date"
-      - "store_name"
       additionalProperties: true
       properties:
-        access_token:
-          type: "string"
-          title: "Access Token"
-          airbyte_secret: true
-          description: "Access Token for making authenticated requests."
-        store_name:
-          type: "string"
-          title: "Store Name"
-          description: "The name of Cart.com Online Store. All API URLs start with\
-            \ https://[mystorename.com]/api/v1/, where [mystorename.com] is the domain\
-            \ name of your store."
+        credentials:
+          title: "Authorization Method"
+          description: ""
+          type: "object"
+          oneOf:
+          - title: "Central API Router"
+            type: "object"
+            order: 0
+            required:
+            - "auth_type"
+            - "user_name"
+            - "user_secret"
+            - "site_id"
+            properties:
+              auth_type:
+                type: "string"
+                const: "CENTRAL_API_ROUTER"
+                order: 0
+              user_name:
+                type: "string"
+                title: "User Name"
+                description: "Enter your application's User Name"
+                airbyte_secret: true
+                order: 1
+              user_secret:
+                type: "string"
+                title: "User Secret"
+                description: "Enter your application's User Secret"
+                airbyte_secret: true
+                order: 2
+              site_id:
+                type: "string"
+                title: "Site ID"
+                description: "You can determine a site provisioning site Id by hitting\
+                  \ https://site.com/store/sitemonitor.aspx and reading the response\
+                  \ param PSID"
+                airbyte_secret: true
+                order: 3
+          - title: "Single Store Access Token"
+            type: "object"
+            order: 1
+            required:
+            - "auth_type"
+            - "access_token"
+            - "store_name"
+            properties:
+              auth_type:
+                type: "string"
+                const: "SINGLE_STORE_ACCESS_TOKEN"
+                order: 0
+              access_token:
+                type: "string"
+                title: "Access Token"
+                airbyte_secret: true
+                order: 1
+                description: "Access Token for making authenticated requests."
+              store_name:
+                type: "string"
+                title: "Store Name"
+                order: 2
+                description: "The name of Cart.com Online Store. All API URLs start\
+                  \ with https://[mystorename.com]/api/v1/, where [mystorename.com]\
+                  \ is the domain name of your store."
         start_date:
           title: "Start Date"
           type: "string"

airbyte-integrations/connectors/source-cart/Dockerfile

@@ -21,5 +21,5 @@ COPY source_cart ./source_cart
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.6
+LABEL io.airbyte.version=0.2.0
 LABEL io.airbyte.name=airbyte/source-cart

airbyte-integrations/connectors/source-cart/acceptance-test-config.yml

@@ -4,6 +4,8 @@ connector_image: airbyte/source-cart:dev
 tests:
   spec:
     - spec_path: "source_cart/spec.json"
+      backward_compatibility_tests_config:
+        disable_for_version: "0.1.6" 
   connection:
     - config_path: "secrets/config.json"
       status: "succeed"
@@ -12,16 +14,26 @@ tests:
       timeout_seconds: 180
   discovery:
     - config_path: "secrets/config.json"
+      backward_compatibility_tests_config:
+        disable_for_version: "0.1.6" 
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
       timeout_seconds: 1800
   incremental:
+    - config_path: "secrets/config_central_api_router.json"
+      configured_catalog_path: "integration_tests/configured_catalog_wo_order_statuses.json"
+      future_state_path: "integration_tests/abnormal_state.json"
+      timeout_seconds: 1800
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
       future_state_path: "integration_tests/abnormal_state.json"
       timeout_seconds: 1800
   full_refresh:
+    - config_path: "secrets/config_central_api_router.json"
+      configured_catalog_path: "integration_tests/configured_catalog_wo_order_statuses.json"
+      timeout_seconds: 1800
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
       timeout_seconds: 1800
+

airbyte-integrations/connectors/source-cart/integration_tests/configured_catalog_wo_order_statuses.json

@@ -0,0 +1,76 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "customers_cart",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "orders",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "addresses",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "order_payments",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "order_items",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "products",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["updated_at"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["updated_at"],
+      "destination_sync_mode": "append"
+    }
+  ]
+}

airbyte-integrations/connectors/source-cart/integration_tests/invalid_config.json

@@ -1,5 +1,8 @@
 {
-  "access_token": "1234567890absc",
-  "start_date": "2021-01-01T00:00:00Z",
-  "store_name": "www.my_pretyy_store.com"
+  "start_date": "2022-01-01T00:00:00Z",
+  "credentials": {
+      "auth_type": "SINGLE_STORE_ACCESS_TOKEN",
+      "store_name": "www.my_pretyy_store.com",
+      "access_token": "1234567890absc"
+  }
 }

airbyte-integrations/connectors/source-cart/source_cart/__init__.py

@@ -22,6 +22,6 @@
 SOFTWARE.
 """
 
-from .source import SourceCart
+from .source import CentralAPIHeaderAuthenticator, CustomHeaderAuthenticator, SourceCart
 
-__all__ = ["SourceCart"]
+__all__ = ["SourceCart", "CentralAPIHeaderAuthenticator", "CustomHeaderAuthenticator"]

airbyte-integrations/connectors/source-cart/source_cart/source.py

@@ -2,6 +2,12 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
+import base64
+import codecs
+import hashlib
+import hmac
+import urllib.parse
+from enum import Enum
 from functools import wraps
 from typing import Any, List, Mapping, Tuple
 
@@ -17,13 +23,65 @@
 from .streams import Addresses, CustomersCart, OrderItems, OrderPayments, Orders, OrderStatuses, Products
 
 
+class AuthMethod(Enum):
+    CENTRAL_API_ROUTER = 1
+    SINGLE_STORE_ACCESS_TOKEN = 2
+
+
 class CustomHeaderAuthenticator(HttpAuthenticator):
-    def __init__(self, access_token):
+    def __init__(self, access_token, store_name):
+        self.auth_method = AuthMethod.SINGLE_STORE_ACCESS_TOKEN
+        self._store_name = store_name
         self._access_token = access_token
 
     def get_auth_header(self) -> Mapping[str, Any]:
         return {"X-AC-Auth-Token": self._access_token}
 
+    def url_base(self) -> str:
+        return f"https://{self._store_name}/api/v1/"
+
+    def extra_params(self, stream, params):
+        return {}
+
+
+class CentralAPIHeaderAuthenticator(HttpAuthenticator):
+    def __init__(self, user_name, user_secret, site_id):
+        self.auth_method = AuthMethod.CENTRAL_API_ROUTER
+        self.user_name = user_name
+        self.user_secret = user_secret
+        self.site_id = site_id
+
+    def get_auth_header(self) -> Mapping[str, Any]:
+        """
+        This method is not implemented here because for the Central API Router
+        needs to build the header for each request based
+        on path + parameters (next token, pagination, page size)
+        To solve this the logic was moved to `request_headers` in CartStream class.
+        """
+        return {}
+
+    def url_base(self) -> str:
+        return "https://public.americommerce.com/api/v1/"
+
+    def extra_params(self, stream, params):
+        return self.generate_auth_signature(stream, params)
+
+    def generate_auth_signature(self, stream, params) -> Mapping[str, Any]:
+        """
+        How to build signature:
+        1. build a string concatenated with:
+            request method (uppercase) & request path and query & provisioning user name
+                example: GET&/api/v1/customers&myUser
+        2. Generate HMACSHA256 hash using this string as the input, and the provisioning user secret as the key
+        3. Base64 this hash to be used as the final value in the header
+        """
+        path_with_params = f"/api/v1/{stream.path()}?{urllib.parse.urlencode(params)}"
+        msg = codecs.encode(f"GET&{path_with_params}&{self.user_name}")
+        key = codecs.encode(self.user_secret)
+        dig = hmac.new(key=key, msg=msg, digestmod=hashlib.sha256).digest()
+        auth_signature = base64.b64encode(dig).decode()
+        return {"X-AC-PUB-Site-ID": self.site_id, "X-AC-PUB-User": self.user_name, "X-AC-PUB-Auth-Signature": auth_signature}
+
 
 class SourceCart(AbstractSource):
     def validate_config_values(func):
@@ -48,12 +106,26 @@ def decorator(self_, *args, **kwargs):
 
         return decorator
 
+    def get_auth(self, config):
+        credentials = config.get("credentials", {})
+        auth_method = credentials.get("auth_type")
+
+        if auth_method == AuthMethod.CENTRAL_API_ROUTER.name:
+            authenticator = CentralAPIHeaderAuthenticator(
+                user_name=credentials["user_name"], user_secret=credentials["user_secret"], site_id=credentials["site_id"]
+            )
+        elif auth_method == AuthMethod.SINGLE_STORE_ACCESS_TOKEN.name:
+            authenticator = CustomHeaderAuthenticator(access_token=credentials["access_token"], store_name=credentials["store_name"])
+        else:
+            raise NotImplementedError(f"Authentication method: {auth_method} not implemented.")
+
+        return authenticator
+
     @validate_config_values
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
         try:
-            authenticator = CustomHeaderAuthenticator(access_token=config["access_token"])
-
-            stream = Products(authenticator=authenticator, start_date=config["start_date"], store_name=config["store_name"])
+            authenticator = self.get_auth(config)
+            stream = Products(authenticator=authenticator, start_date=config["start_date"])
             records = stream.read_records(sync_mode=SyncMode.full_refresh)
             next(records)
             return True, None
@@ -67,11 +139,10 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
 
     @validate_config_values
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
-        authenticator = CustomHeaderAuthenticator(access_token=config["access_token"])
+        authenticator = self.get_auth(config)
         args = {
             "authenticator": authenticator,
             "start_date": config["start_date"],
-            "store_name": config["store_name"],
             "end_date": config.get("end_date"),
         }
         return [