HGI-11002: Add error classification to raise correct errors

tap_airwallex/client.py

@@ -8,7 +8,10 @@
 from memoization import cached
 
 from hotglue_singer_sdk.streams import RESTStream
-from hotglue_singer_sdk.exceptions import FatalAPIError
+from hotglue_singer_sdk.exceptions import FatalAPIError, RetriableAPIError
+from hotglue_etl_exceptions import InsufficientPermissionsError, InvalidCredentialsError
+
+from tap_airwallex.exceptions import is_permission_error
 
 from tap_airwallex.auth import AirwallexAuthenticator
 
@@ -78,6 +81,12 @@ def validate_response(self, response: requests.Response) -> None:
                 response.url,
                 response.text,
             )
+        if response.status_code in (401, 403):
+            if is_permission_error(response):
+                raise InsufficientPermissionsError(self.response_error_message(response))
+            raise InvalidCredentialsError(self.response_error_message(response))
+        if response.status_code == 429:
+            raise RetriableAPIError(self.response_error_message(response), response)
         super().validate_response(response)
 
     def prepare_request(

tap_airwallex/exceptions.py

@@ -0,0 +1,21 @@
+"""Airwallex-specific exceptions."""
+
+import requests
+
+# Airwallex returns 401 for both bad credentials and insufficient scope,
+#   bad key gives {"code": "credentials_invalid", "message": "UNAUTHORIZED"}
+#   bad scope gives {"code": "unauthorized", "message": "Insufficient permissions"}
+PERMISSION_ERROR_HINTS = ("permission", "scope")
+
+
+def is_permission_error(response: requests.Response) -> bool:
+    try:
+        body = response.json()
+    except ValueError:
+        return False
+    if not isinstance(body, dict):
+        return False
+    text = " ".join(
+        str(body.get(field, "")) for field in ("code", "message")
+    ).lower()
+    return any(hint in text for hint in PERMISSION_ERROR_HINTS)

tap_airwallex/tap.py

@@ -3,7 +3,9 @@
 from typing import List
 
 from hotglue_singer_sdk import Tap, Stream
-from hotglue_singer_sdk import typing as th  
+from hotglue_singer_sdk import typing as th
+from hotglue_singer_sdk.helpers.capabilities import AlertingLevel
+from hotglue_etl_exceptions import InsufficientPermissionsError, InvalidCredentialsError
 from tap_airwallex.streams import (
     AccountsStream,
     AccountDetailsStream,
@@ -38,6 +40,12 @@ class TapAirwallex(Tap):
     """airwallex tap class."""
     name = "tap-airwallex"
 
+    alerting_level = AlertingLevel.ERROR
+    exception_alerting_level_map = {
+        InvalidCredentialsError: AlertingLevel.NONE,
+        InsufficientPermissionsError: AlertingLevel.NONE,
+    }
+
     config_jsonschema = th.PropertiesList(
         th.Property(
             "api_key",