✨ Source Shopify: add resiliency on some transient errors using the HttpClient

airbyte-integrations/connectors/source-shopify/metadata.yaml

@@ -11,7 +11,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 9da77001-af33-4bcd-be46-6252bf9342b9
-  dockerImageTag: 2.1.4
+  dockerImageTag: 2.1.5
   dockerRepository: airbyte/source-shopify
   documentationUrl: https://docs.airbyte.com/integrations/sources/shopify
   githubIssueLabel: source-shopify

airbyte-integrations/connectors/source-shopify/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "2.1.4"
+version = "2.1.5"
 name = "source-shopify"
 description = "Source CDK implementation for Shopify."
 authors = [ "Airbyte <[email protected]>",]
@@ -28,3 +28,4 @@ source-shopify = "source_shopify.run:run"
 requests-mock = "^1.11.0"
 pytest-mock = "^3.12.0"
 pytest = "^8.0.0"
+freezegun = "^1.4.0"

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

@@ -83,16 +83,14 @@ class ShopifyScopes:
     logger = logging.getLogger("airbyte")
 
     def __init__(self, config: Mapping[str, Any]) -> None:
+        self.permitted_streams: List[str] = list(ALWAYS_PERMITTED_STREAMS)
+        self.not_permitted_streams: List[set[str, str]] = []
         self.user_scopes = self.get_user_scopes(config)
         # for each stream check the authenticated user has all scopes required
         self.get_streams_from_user_scopes()
         # log if there are streams missing scopes and should be omitted
         self.emit_missing_scopes()
 
-    # the list of validated streams
-    permitted_streams: List[str] = ALWAYS_PERMITTED_STREAMS
-    # the list of not permitted streams
-    not_permitted_streams: List[set[str, str]] = []
     # template for the log message
     missing_scope_message: str = (
         "The stream `{stream}` could not be synced without the `{scope}` scope. Please check the `{scope}` is granted."

airbyte-integrations/connectors/source-shopify/unit_tests/integration/api/authentication.py

@@ -0,0 +1,83 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+import json
+
+from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
+from airbyte_cdk.test.mock_http.request import ANY_QUERY_PARAMS
+from airbyte_cdk.test.mock_http.response_builder import find_template
+
+_ALL_SCOPES = [
+    "read_all_cart_transforms",
+    "read_all_checkout_completion_target_customizations",
+    "read_all_orders",
+    "read_analytics",
+    "read_assigned_fulfillment_orders",
+    "read_cart_transforms",
+    "read_channels",
+    "read_companies",
+    "read_content",
+    "read_custom_fulfillment_services",
+    "read_customer_data_erasure",
+    "read_customer_merge",
+    "read_customers",
+    "read_dery_customizations",
+    "read_discounts",
+    "read_draft_orders",
+    "read_files",
+    "read_fulfillment_constraint_rules",
+    "read_fulfillments",
+    "read_gates",
+    "read_gdpr_data_request",
+    "read_gift_cards",
+    "read_inventory",
+    "read_legal_policies",
+    "read_locales",
+    "read_locations",
+    "read_marketing_events",
+    "read_markets",
+    "read_merchant_managed_fulfillment_orders",
+    "read_online_store_navigation",
+    "read_online_store_pages",
+    "read_order_edits",
+    "read_order_submission_rules",
+    "read_orders",
+    "read_packing_slip_templates",
+    "read_payment_customizations",
+    "read_payment_terms",
+    "read_pixels",
+    "read_price_rules",
+    "read_product_feeds",
+    "read_product_listings",
+    "read_products",
+    "read_publications",
+    "read_purchase_options",
+    "read_reports",
+    "read_resource_feedbacks",
+    "read_returns",
+    "read_script_tags",
+    "read_shipping",
+    "read_shopify_credit",
+    "read_shopify_payments_accounts",
+    "read_shopify_payments_bank_accounts",
+    "read_shopify_payments_disputes",
+    "read_shopify_payments_payouts",
+    "read_shopify_payments_provider_accounts_sensitive",
+    "read_store_credit_account_transactions",
+    "read_themes",
+    "read_third_party_fulfillment_orders",
+    "read_translations"
+]
+
+
+def set_up_shop(http_mocker: HttpMocker, shop_name: str) -> None:
+    http_mocker.get(
+        HttpRequest(f"https://{shop_name}.myshopify.com/admin/api/2024-04/shop.json", query_params=ANY_QUERY_PARAMS),
+        HttpResponse(json.dumps(find_template("shop", __file__)), status_code=200),
+    )
+
+
+def grant_all_scopes(http_mocker: HttpMocker, shop_name: str) -> None:
+    http_mocker.get(
+        HttpRequest(f"https://{shop_name}.myshopify.com/admin/oauth/access_scopes.json"),
+        HttpResponse(json.dumps({"access_scopes": [{"handle": scope} for scope in _ALL_SCOPES]}), status_code=200),
+    )

airbyte-integrations/connectors/source-shopify/unit_tests/integration/api/bulk.py

@@ -0,0 +1,177 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+import json
+from datetime import datetime
+from random import randint
+
+from airbyte_cdk.test.mock_http import HttpRequest, HttpResponse
+
+
+def _create_job_url(shop_name: str) -> str:
+    return f"https://{shop_name}.myshopify.com/admin/api/2024-04/graphql.json"
+
+
+def create_job_creation_request(shop_name: str, lower_boundary: datetime, upper_boundary: datetime) -> HttpRequest:
+    outer_query = """mutation {
+                bulkOperationRunQuery(
+                    query: \"\"\"
+                     %INNER_QUERY_TOKEN%
+                    \"\"\"
+                ) {
+                    bulkOperation {
+                        id
+                        status
+                        createdAt
+                    }
+                    userErrors {
+                        field
+                        message
+                    }
+                }
+            }"""
+    inner_query = """{
+  orders(
+    query: \"updated_at:>='%LOWER_BOUNDARY_TOKEN%' AND updated_at:<='%UPPER_BOUNDARY_TOKEN%'\"
+    sortKey: UPDATED_AT
+  ) {
+    edges {
+      node {
+        __typename
+        id
+        metafields {
+          edges {
+            node {
+              __typename
+              id
+              namespace
+              value
+              key
+              description
+              createdAt
+              updatedAt
+              type
+            }
+          }
+        }
+      }
+    }
+  }
+}"""
+    inner_query = inner_query.replace("%LOWER_BOUNDARY_TOKEN%", lower_boundary.isoformat())
+    inner_query = inner_query.replace("%UPPER_BOUNDARY_TOKEN%", upper_boundary.isoformat())
+    outer_query = outer_query.replace("%INNER_QUERY_TOKEN%", inner_query)
+
+    return HttpRequest(
+        url=_create_job_url(shop_name),
+        body=json.dumps({"query": outer_query})
+    )
+
+
+def create_job_status_request(shop_name: str, job_id: str) -> HttpRequest:
+    return HttpRequest(
+        url=_create_job_url(shop_name),
+        body=f"""query {{
+                    node(id: "{job_id}") {{
+                        ... on BulkOperation {{
+                            id
+                            status
+                            errorCode
+                            createdAt
+                            objectCount
+                            fileSize
+                            url
+                            partialDataUrl
+                        }}
+                    }}
+                }}"""
+    )
+
+
+class JobCreationResponseBuilder:
+    def __init__(self) -> None:
+        self._template = {
+            "data": {
+                "bulkOperationRunQuery": {
+                    "bulkOperation": {
+                        "id": "gid://shopify/BulkOperation/0",
+                        "status": "CREATED",
+                        "createdAt": "2024-05-05T02:00:00Z"
+                    },
+                    "userErrors": []
+                }
+            },
+            "extensions": {
+                "cost": {
+                    "requestedQueryCost": 10,
+                    "actualQueryCost": 10,
+                    "throttleStatus": {
+                        "maximumAvailable": 2000.0,
+                        "currentlyAvailable": 1990,
+                        "restoreRate": 100.0
+                    }
+                }
+            }
+        }
+
+    def with_bulk_operation_id(self, bulk_operation_id: str) -> "JobCreationResponseBuilder":
+        self._template["data"]["bulkOperationRunQuery"]["bulkOperation"]["id"] = bulk_operation_id
+        return self
+
+    def build(self) -> HttpResponse:
+        return HttpResponse(json.dumps(self._template), status_code=200)
+
+
+class JobStatusResponseBuilder:
+    def __init__(self) -> None:
+        self._template = {
+            "data": {
+                "node": {},
+                "extensions": {
+                    "cost": {
+                        "requestedQueryCost": 1,
+                        "actualQueryCost": 1,
+                        "throttleStatus": {
+                            "maximumAvailable": 2000.0,
+                            "currentlyAvailable": 1999,
+                            "restoreRate": 100.0
+                        }
+                    }
+                }
+            }
+        }
+
+    def with_completed_status(self, bulk_operation_id: str, job_result_url: str) -> "JobStatusResponseBuilder":
+        self._template["data"]["node"] = {
+            "id": bulk_operation_id,
+            "status": "COMPLETED",
+            "errorCode": None,
+            "createdAt": "2024-05-05T00:45:48Z",
+            "objectCount": "4",
+            "fileSize": "774",
+            "url": job_result_url,
+            "partialDataUrl": None
+        }
+        return self
+
+    def build(self) -> HttpResponse:
+        return HttpResponse(json.dumps(self._template), status_code=200)
+
+
+class MetafieldOrdersJobResponseBuilder:
+    def __init__(self) -> None:
+        self._records = []
+
+    def _any_record(self) -> str:
+        an_id = str(randint(1000000000000, 9999999999999))
+        a_parent_id = str(randint(1000000000000, 9999999999999))
+        return f"""{{"__typename":"Order","id":"gid:\/\/shopify\/Order\/{a_parent_id}"}}
+{{"__typename":"Metafield","id":"gid:\/\/shopify\/Metafield\/{an_id}","namespace":"my_fields","value":"asdfasdf","key":"purchase_order","description":null,"createdAt":"2023-04-13T12:09:50Z","updatedAt":"2024-05-05T01:09:50Z","type":"single_line_text_field","__parentId":"gid:\/\/shopify\/Order\/{a_parent_id}"}}
+"""
+
+
+    def with_record(self) -> "MetafieldOrdersJobResponseBuilder":
+        self._records.append(self._any_record())
+        return self
+
+    def build(self) -> HttpResponse:
+        return HttpResponse("".join(self._records), status_code=200)