Source Cart.com: implement Central API Router access method and improve backoff policy (#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 <octavia-squidington-iii@users.noreply.github.com>

Author: marcosmarxm

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 [