🐛 Source Shopify: Fixed rate limits throttling

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/9da77001-af33-4bcd-be46-6252bf9342b9.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "9da77001-af33-4bcd-be46-6252bf9342b9",
   "name": "Shopify",
   "dockerRepository": "airbyte/source-shopify",
-  "dockerImageTag": "0.1.12",
+  "dockerImageTag": "0.1.13",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/shopify"
 }

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

@@ -148,7 +148,7 @@
 - sourceDefinitionId: 9da77001-af33-4bcd-be46-6252bf9342b9
   name: Shopify
   dockerRepository: airbyte/source-shopify
-  dockerImageTag: 0.1.12
+  dockerImageTag: 0.1.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/shopify
 - sourceDefinitionId: e87ffa8e-a3b5-f69c-9076-6011339de1f6
   name: Redshift

airbyte-integrations/connectors/source-shopify/Dockerfile

@@ -1,16 +1,32 @@
-FROM python:3.7-slim
+FROM python:3.7.11-alpine3.14 as base
 
-# Bash is installed for more convenient debugging.
-RUN apt-get update && apt-get install -y bash && rm -rf /var/lib/apt/lists/*
+# build and load all requirements 
+FROM base as builder
+WORKDIR /airbyte/integration_code
+
+# upgrade pip to the latest version
+RUN apk --no-cache upgrade && pip install --upgrade pip
 
+COPY setup.py ./
+# install necessary packages to a temporary folder 
+RUN pip install --prefix=/install .
+
+# build a clean environment  
+FROM base
 WORKDIR /airbyte/integration_code
-COPY source_shopify ./source_shopify
+
+# copy all loaded and built libraries to a pure basic image
+COPY --from=builder /install /usr/local
+
+# Bash is installed for more convenient debugging.
+RUN apk --no-cache add bash
+
+# copy payload code only
 COPY main.py ./
-COPY setup.py ./
-RUN pip install .
+COPY source_shopify ./source_shopify
 
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.12
+LABEL io.airbyte.version=0.1.13
 LABEL io.airbyte.name=airbyte/source-shopify

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

@@ -1,5 +1,5 @@
 {
   "shop": "SHOP_NAME",
-  "api_key": "API_KEY",
+  "start_date": "2020-11-01",
   "api_password": "API_PASSWORD"
 }

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/custom_collections.json

@@ -22,7 +22,8 @@
       "type": ["null", "string"]
     },
     "updated_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "image": {
       "properties": {
@@ -36,7 +37,8 @@
           "type": ["null", "integer"]
         },
         "created_at": {
-          "type": ["null", "string"]
+          "type": ["null", "string"],
+          "format": "date-time"
         },
         "height": {
           "type": ["null", "integer"]
@@ -45,7 +47,8 @@
       "type": ["null", "object"]
     },
     "published_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "template_suffix": {
       "type": ["null", "string"]

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/draft_orders.json

@@ -559,8 +559,7 @@
         "discounted_price_set": {},
         "price_set": {},
         "price": {
-          "type": ["null", "string"],
-          "format": "singer.decimal"
+          "type": ["null", "string"]
         },
         "title": {
           "type": ["null", "string"]
@@ -663,8 +662,7 @@
       "type": ["null", "boolean"]
     },
     "total_price": {
-      "type": ["null", "string"],
-      "format": "singer.decimal"
+      "type": ["null", "string"]
     },
     "completed_at": {
       "type": ["null", "string"],

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/order_refunds.json

@@ -14,15 +14,13 @@
             "type": ["null", "integer"]
           },
           "tax_amount": {
-            "type": ["null", "number"],
-            "multipleOf": 1e-10
+            "type": ["null", "number"]
           },
           "refund_id": {
             "type": ["null", "integer"]
           },
           "amount": {
-            "type": ["null", "number"],
-            "multipleOf": 1e-10
+            "type": ["null", "number"]
           },
           "kind": {
             "type": ["null", "string"]

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/orders.json

@@ -55,7 +55,8 @@
       }
     },
     "closed_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "confirmed": {
       "type": ["null", "boolean"]
@@ -64,7 +65,8 @@
       "type": ["null", "string"]
     },
     "created_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "currency": {
       "type": ["null", "string"]
@@ -533,7 +535,8 @@
       "type": ["null", "integer"]
     },
     "updated_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "user_id": {
       "type": ["null", "number"]
@@ -735,7 +738,8 @@
             "type": ["null", "string"]
           },
           "created_at": {
-            "type": ["null", "string"]
+            "type": ["null", "string"],
+            "format": "date-time"
           },
           "location_id": {
             "type": ["null", "integer"]
@@ -774,7 +778,8 @@
             "type": ["null", "array"]
           },
           "updated_at": {
-            "type": ["null", "string"]
+            "type": ["null", "string"],
+            "format": "date-time"
           },
           "line_items": {
             "type": ["null", "array"],

airbyte-integrations/connectors/source-shopify/source_shopify/schemas/transactions.json

@@ -40,7 +40,8 @@
       "type": ["null", "integer"]
     },
     "created_at": {
-      "type": ["null", "string"]
+      "type": ["null", "string"],
+      "format": "date-time"
     },
     "status": {
       "type": ["null", "string"]

airbyte-integrations/connectors/source-shopify/source_shopify/source.py

@@ -22,6 +22,7 @@
 # SOFTWARE.
 #
 
+
 from abc import ABC, abstractmethod
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Tuple
 from urllib.parse import parse_qsl, urlparse
@@ -32,7 +33,9 @@
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.http import HttpStream
-from airbyte_cdk.sources.streams.http.auth import HttpAuthenticator
+from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
+
+from .utils import BalanceRateLimit as balancer
 
 
 class ShopifyStream(HttpStream, ABC):
@@ -46,11 +49,10 @@ class ShopifyStream(HttpStream, ABC):
     order_field = "updated_at"
     filter_field = "updated_at_min"
 
-    def __init__(self, shop: str, start_date: str, api_password: str, **kwargs):
+    def __init__(self, shop: str, start_date: str, **kwargs):
         super().__init__(**kwargs)
         self.start_date = start_date
         self.shop = shop
-        self.api_password = api_password
 
     @property
     def url_base(self) -> str:
@@ -73,6 +75,7 @@ def request_params(self, next_page_token: Mapping[str, Any] = None, **kwargs) ->
             params[self.filter_field] = self.start_date
         return params
 
+    @balancer.balance_rate_limit()
     def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
         json_response = response.json()
         records = json_response.get(self.data_field, []) if self.data_field is not None else json_response
@@ -125,7 +128,7 @@ class OrderSubstream(IncrementalShopifyStream):
     def read_records(
         self, stream_state: Mapping[str, Any] = None, stream_slice: Optional[Mapping[str, Any]] = None, **kwargs
     ) -> Iterable[Mapping[str, Any]]:
-        orders_stream = Orders(authenticator=self.authenticator, shop=self.shop, start_date=self.start_date, api_password=self.api_password)
+        orders_stream = Orders(authenticator=self.authenticator, shop=self.shop, start_date=self.start_date)
         for data in orders_stream.read_records(sync_mode=SyncMode.full_refresh):
             slice = super().read_records(stream_slice={"order_id": data["id"]}, **kwargs)
             yield from self.filter_records_newer_than_state(stream_state=stream_state, records_slice=slice)
@@ -300,25 +303,20 @@ def read_records(
         self, stream_state: Mapping[str, Any] = None, stream_slice: Optional[Mapping[str, Any]] = None, **kwargs
     ) -> Iterable[Mapping[str, Any]]:
         stream_state = stream_state or {}
-        price_rules_stream = PriceRules(
-            authenticator=self.authenticator, shop=self.shop, start_date=self.start_date, api_password=self.api_password
-        )
+        price_rules_stream = PriceRules(authenticator=self.authenticator, shop=self.shop, start_date=self.start_date)
         for data in price_rules_stream.read_records(sync_mode=SyncMode.full_refresh):
             discount_slice = super().read_records(stream_slice={"price_rule_id": data["id"]}, **kwargs)
             yield from self.filter_records_newer_than_state(stream_state=stream_state, records_slice=discount_slice)
 
 
-class ShopifyAuthenticator(HttpAuthenticator):
+class ShopifyAuthenticator(TokenAuthenticator):
 
     """
     Making Authenticator to be able to accept Header-Based authentication.
     """
 
-    def __init__(self, token: str):
-        self.token = token
-
     def get_auth_header(self) -> Mapping[str, Any]:
-        return {"X-Shopify-Access-Token": f"{self.token}"}
+        return {"X-Shopify-Access-Token": f"{self._token}"}
 
 
 # Basic Connections Check
@@ -328,16 +326,13 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
         """
         Testing connection availability for the connector.
         """
-
-        shop = config["shop"]
-        api_pass = config["api_password"]
+        auth = ShopifyAuthenticator(token=config["api_password"]).get_auth_header()
         api_version = "2021-07"  # Latest Stable Release
 
-        headers = {"X-Shopify-Access-Token": api_pass}
-        url = f"https://{shop}.myshopify.com/admin/api/{api_version}/shop.json"
+        url = f"https://{config['shop']}.myshopify.com/admin/api/{api_version}/shop.json"
 
         try:
-            session = requests.get(url, headers=headers)
+            session = requests.get(url, headers=auth)
             session.raise_for_status()
             return True, None
         except requests.exceptions.RequestException as e:
@@ -351,7 +346,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         """
 
         auth = ShopifyAuthenticator(token=config["api_password"])
-        args = {"authenticator": auth, "shop": config["shop"], "start_date": config["start_date"], "api_password": config["api_password"]}
+        args = {"authenticator": auth, "shop": config["shop"], "start_date": config["start_date"]}
         return [
             Customers(**args),
             Orders(**args),