🐛 Source Shopify: improve error messages, refactored code

airbyte-integrations/connectors/source-shopify/Dockerfile

@@ -28,5 +28,5 @@ 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.5.0
+LABEL io.airbyte.version=0.5.1
 LABEL io.airbyte.name=airbyte/source-shopify

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

@@ -35,6 +35,8 @@ acceptance_tests:
             bypass_reason: The stream holds data up to 1 month then records are removed by Shopify.
           - name: balance_transactions
             bypass_reason: The stream requires real purchases to fill in the data.
+          - name: customer_saved_search
+            bypass_reason: The stream is not available for our sandbox.
   incremental:
     tests:
       - config_path: "secrets/config.json"

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

@@ -2,7 +2,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 9da77001-af33-4bcd-be46-6252bf9342b9
-  dockerImageTag: 0.5.0
+  dockerImageTag: 0.5.1
   dockerRepository: airbyte/source-shopify
   githubIssueLabel: source-shopify
   icon: shopify.svg

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

@@ -2,16 +2,16 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import logging
 from typing import Any, Dict, Mapping
 
-from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator
 
 
 class NotImplementedAuth(Exception):
     """Not implemented Auth option error"""
 
-    logger = AirbyteLogger()
+    logger = logging.getLogger("airbyte")
 
     def __init__(self, auth_method: str = None):
         self.message = f"Not implemented Auth method = {auth_method}"

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

@@ -3,6 +3,7 @@
 #
 
 
+import logging
 from abc import ABC, abstractmethod
 from functools import cached_property
 from typing import Any, Dict, Iterable, List, Mapping, MutableMapping, Optional, Tuple, Union
@@ -13,15 +14,16 @@
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.http import HttpStream
-from requests.exceptions import RequestException
+from requests.exceptions import ConnectionError, InvalidURL, JSONDecodeError, RequestException, SSLError
 
 from .auth import ShopifyAuthenticator
 from .graphql import get_query_products
 from .transform import DataTypeEnforcer
 from .utils import SCOPES_MAPPING, ApiTypeEnum
 from .utils import EagerlyCachedStreamState as stream_state_cache
-from .utils import ErrorAccessScopes
+from .utils import ShopifyAccessScopesError, ShopifyBadJsonError, ShopifyConnectionError, ShopifyNonRetryableErrors
 from .utils import ShopifyRateLimiter as limiter
+from .utils import ShopifyWrongShopNameError
 
 
 class ShopifyStream(HttpStream, ABC):
@@ -34,7 +36,11 @@ class ShopifyStream(HttpStream, ABC):
     order_field = "updated_at"
     filter_field = "updated_at_min"
 
+    # define default logger
+    logger = logging.getLogger("airbyte")
+
     raise_on_http_errors = True
+    max_retries = 5
 
     def __init__(self, config: Dict):
         super().__init__(authenticator=config["authenticator"])
@@ -77,7 +83,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                 records = json_response.get(self.data_field, []) if self.data_field is not None else json_response
                 yield from self.produce_records(records)
             except RequestException as e:
-                self.logger.warn(f"Unexpected error in `parse_ersponse`: {e}, the actual response data: {response.text}")
+                self.logger.warning(f"Unexpected error in `parse_ersponse`: {e}, the actual response data: {response.text}")
                 yield {}
 
     def produce_records(self, records: Union[Iterable[Mapping[str, Any]], Mapping[str, Any]] = None) -> Iterable[Mapping[str, Any]]:
@@ -97,15 +103,11 @@ def produce_records(self, records: Union[Iterable[Mapping[str, Any]], Mapping[st
                 yield self._transformer.transform(record)
 
     def should_retry(self, response: requests.Response) -> bool:
-        error_mapping = {
-            404: f"Stream `{self.name}` - not available or missing, skipping...",
-            403: f"Stream `{self.name}` - insufficient permissions, skipping...",
-            # extend the mapping with more handable errors, if needed.
-        }
+        known_errors = ShopifyNonRetryableErrors(self.name)
         status = response.status_code
-        if status in error_mapping.keys():
+        if status in known_errors.keys():
             setattr(self, "raise_on_http_errors", False)
-            self.logger.warn(error_mapping.get(status))
+            self.logger.warning(known_errors.get(status))
             return False
         else:
             return super().should_retry(response)
@@ -162,13 +164,13 @@ def filter_records_newer_than_state(self, stream_state: Mapping[str, Any] = None
                             yield record
                     else:
                         # old entities could have cursor field in place, but set to null
-                        self.logger.warn(
+                        self.logger.warning(
                             f"Stream `{self.name}`, Record ID: `{record.get(self.primary_key)}` cursor value is: {record_value}, record is emitted without state comparison"
                         )
                         yield record
                 else:
                     # old entities could miss the cursor field
-                    self.logger.warn(
+                    self.logger.warning(
                         f"Stream `{self.name}`, Record ID: `{record.get(self.primary_key)}` missing cursor field: {self.cursor_field}, record is emitted without state comparison"
                     )
                     yield record
@@ -447,7 +449,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                 json_response = response.json()["data"]["products"]["nodes"]
                 yield from self.produce_records(json_response)
             except RequestException as e:
-                self.logger.warn(f"Unexpected error in `parse_ersponse`: {e}, the actual response data: {response.text}")
+                self.logger.warning(f"Unexpected error in `parse_ersponse`: {e}, the actual response data: {response.text}")
                 yield {}
 
 
@@ -781,20 +783,53 @@ def path(self, **kwargs) -> str:
         return f"{self.data_field}.json"
 
 
+class ConnectionCheckTest:
+    def __init__(self, config: Mapping[str, Any]):
+        self.config = config
+        # use `Shop` as a test stream for connection check
+        self.test_stream = Shop(self.config)
+        # setting `max_retries` to 0 for the stage of `check connection`,
+        # because it keeps retrying for wrong shop names,
+        # but it should stop immediately
+        self.test_stream.max_retries = 0
+
+    def describe_error(self, pattern: str, shop_name: str = None, details: Any = None, **kwargs) -> str:
+        connection_check_errors_map: Mapping[str, Any] = {
+            "connection_error": f"Connection could not be established using `Shopify Store`: {shop_name}. Make sure it's valid and try again.",
+            "request_exception": f"Request was not successfull, check your `input configuation` and try again. Details: {details}",
+            "index_error": f"Failed to access the Shopify store `{shop_name}`. Verify the entered Shopify store or API Key in `input configuration`.",
+            # add the other patterns and description, if needed...
+        }
+        return connection_check_errors_map.get(pattern)
+
+    def test_connection(self) -> tuple[bool, str]:
+        shop_name = self.config.get("shop")
+        if not shop_name:
+            return False, "The `Shopify Store` name is missing. Make sure it's entered and valid."
+
+        try:
+            response = list(self.test_stream.read_records(sync_mode=None))
+            # check for the shop_id is present in the response
+            shop_id = response[0].get("id")
+            if shop_id is not None:
+                return True, None
+            else:
+                return False, f"The `shop_id` is invalid: {shop_id}"
+        except (SSLError, ConnectionError):
+            return False, self.describe_error("connection_error", shop_name)
+        except RequestException as req_error:
+            return False, self.describe_error("request_exception", details=req_error)
+        except IndexError:
+            return False, self.describe_error("index_error", shop_name, response)
+
+
 class SourceShopify(AbstractSource):
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, any]:
         """
         Testing connection availability for the connector.
         """
         config["authenticator"] = ShopifyAuthenticator(config)
-        try:
-            response = list(Shop(config).read_records(sync_mode=None))
-            # check for the shop_id is present in the response
-            shop_id = response[0].get("id")
-            if shop_id is not None:
-                return True, None
-        except (requests.exceptions.RequestException, IndexError) as e:
-            return False, e
+        return ConnectionCheckTest(config).test_connection()
 
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         """
@@ -866,12 +901,21 @@ def get_user_scopes(config):
         session = requests.Session()
         url = f"https://{config['shop']}.myshopify.com/admin/oauth/access_scopes.json"
         headers = config["authenticator"].get_auth_header()
-        response = session.get(url, headers=headers).json()
-        access_scopes = response.get("access_scopes")
+
+        try:
+            response = session.get(url, headers=headers).json()
+            access_scopes = response.get("access_scopes")
+        except InvalidURL:
+            raise ShopifyWrongShopNameError(url)
+        except JSONDecodeError as json_error:
+            raise ShopifyBadJsonError(json_error)
+        except (SSLError, ConnectionError) as con_error:
+            raise ShopifyConnectionError(con_error)
+
         if access_scopes:
             return access_scopes
         else:
-            raise ErrorAccessScopes(f"Reason: {response}")
+            raise ShopifyAccessScopesError(response)
 
     @staticmethod
     def format_name(name):

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

@@ -6,7 +6,7 @@
 import enum
 from functools import wraps
 from time import sleep
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional
 
 import requests
 
@@ -49,13 +49,51 @@
 }
 
 
-class ErrorAccessScopes(Exception):
+class ShopifyNonRetryableErrors:
+    """Holds the errors clasification and messaging scenarios."""
+
+    def __new__(self, stream: str) -> Mapping[str, Any]:
+        return {
+            401: f"Stream `{stream}`. Failed to access the Shopify store with provided API token. Verify your API token is valid.",
+            402: f"Stream `{stream}`. The shop's plan does not have access to this feature. Please upgrade your plan to be  able to access this stream.",
+            403: f"Stream `{stream}`. Unable to access Shopify endpoint for {stream}. Check that you have the appropriate access scopes to read data from this endpoint.",
+            404: f"Stream `{stream}`. Not available or missing.",
+            # extend the mapping with more handable errors, if needed.
+        }
+
+
+class ShopifyAccessScopesError(Exception):
     """Raises the error if authenticated user doesn't have access to verify the grantted scopes."""
 
     help_url = "https://shopify.dev/docs/api/usage/access-scopes#authenticated-access-scopes"
 
+    def __init__(self, response):
+        super().__init__(
+            f"Reason: Scopes are not available, make sure you're using the correct `Shopify Store` name. Actual response: {response}. More info about: {self.help_url}"
+        )
+
+
+class ShopifyBadJsonError(ShopifyAccessScopesError):
+    """Raises the error when Shopify replies with broken json for `access_scopes` request"""
+
     def __init__(self, message):
-        super().__init__(f"{message}. More info about: {self.help_url}")
+        super().__init__(f"Reason: Bad JSON Response from the Shopify server. Details: {message}.")
+
+
+class ShopifyConnectionError(ShopifyAccessScopesError):
+    """Raises the error when Shopify replies with broken connection error for `access_scopes` request"""
+
+    def __init__(self, details):
+        super().__init__(f"Invalid `Shopify Store` name used or `host` couldn't be verified by Shopify. Details: {details}")
+
+
+class ShopifyWrongShopNameError(Exception):
+    """Raises the error when `Shopify Store` name is incorrect or couldn't be verified by the Shopify"""
+
+    def __init__(self, url):
+        super().__init__(
+            f"Reason: The `Shopify Store` name is invalid or missing for `input configuration`, make sure it's valid. Details: {url}"
+        )
 
 
 class UnrecognisedApiType(Exception):

docs/integrations/sources/shopify.md

@@ -149,6 +149,7 @@ This is expected when the connector hits the 429 - Rate Limit Exceeded HTTP Erro
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                         |
 | :------ | :--------- | :------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------ |
+| 0.5.1   | 2023-07-13 | [28700](https://github.com/airbytehq/airbyte/pull/28700) | Improved `error messages` with more user-friendly description, refactored code  |
 | 0.5.0   | 2023-06-13 | [27732](https://github.com/airbytehq/airbyte/pull/27732) | License Update: Elv2                                                                                                            |
 | 0.4.0   | 2023-06-13 | [27083](https://github.com/airbytehq/airbyte/pull/27083) | Added `CustomerSavedSearch`, `CustomerAddress` and `Countries` streams                                                          |
 | 0.3.4   | 2023-05-10 | [25961](https://github.com/airbytehq/airbyte/pull/25961) | Added validation for `shop` in input configuration (accepts non-url-like inputs)                                                |