feat: improved the exception handling to ensure that they are aligned across all SDKs and that API errors can be differentiated

Author: mridang

spec/auth/using_access_token_spec.py

@@ -3,7 +3,7 @@
 import pytest
 
 import zitadel_client as zitadel
-from zitadel_client.exceptions import OpenApiError
+from zitadel_client.exceptions import ZitadelError
 
 
 @pytest.fixture(scope="module")
@@ -58,5 +58,5 @@ def test_raises_api_exception_with_invalid_token(
             base_url,
             "invalid",
         )
-        with pytest.raises(OpenApiError):
+        with pytest.raises(ZitadelError):
             client.settings.settings_service_get_general_settings()

spec/auth/using_client_credentials_spec.py

@@ -3,7 +3,7 @@
 import pytest
 
 import zitadel_client as zitadel
-from zitadel_client.exceptions import OpenApiError
+from zitadel_client.exceptions import ZitadelError
 
 
 @pytest.fixture(scope="module")
@@ -70,5 +70,5 @@ def test_raises_api_exception_with_invalid_client_credentials(
             "invalid",
             "invalid",
         )
-        with pytest.raises(OpenApiError):
+        with pytest.raises(ZitadelError):
             client.settings.settings_service_get_general_settings()

spec/auth/using_private_key_spec.py

@@ -4,7 +4,7 @@
 import pytest
 
 import zitadel_client as zitadel
-from zitadel_client import OpenApiError
+from zitadel_client import ZitadelError
 
 
 @pytest.fixture(scope="module")
@@ -58,5 +58,5 @@ def test_raises_api_exception_with_invalid_private_key(
             "https://zitadel.cloud",
             key_file,
         )
-        with pytest.raises(OpenApiError):
+        with pytest.raises(ZitadelError):
             client.settings.settings_service_get_general_settings()

spec/check_session_service_spec.py

@@ -5,7 +5,7 @@
 import pytest
 
 import zitadel_client as zitadel
-from zitadel_client.exceptions import ApiException
+from zitadel_client.exceptions import ApiError
 from zitadel_client.models import (
     SessionServiceChecks,
     SessionServiceCheckUser,
@@ -61,7 +61,7 @@ def session(client: zitadel.Zitadel) -> Generator[SessionServiceCreateSessionRes
             response.session_id if response.session_id is not None else "",
             delete_body,
         )
-    except ApiException:
+    except ApiError:
         pass
 
 
@@ -124,7 +124,7 @@ def test_raises_api_exception_for_nonexistent_session(
         session: SessionServiceCreateSessionResponse,
     ) -> None:
         """Raises an ApiException when retrieving a non-existent session."""
-        with pytest.raises(ApiException):
+        with pytest.raises(ApiError):
             client.sessions.session_service_get_session(
                 str(uuid.uuid4()),
                 session_token=session.session_token,

spec/check_user_service_spec.py

@@ -5,7 +5,7 @@
 import pytest
 
 import zitadel_client as zitadel
-from zitadel_client.exceptions import ApiException
+from zitadel_client.exceptions import ApiError
 from zitadel_client.models import (
     UserServiceAddHumanUserRequest,
     UserServiceAddHumanUserResponse,
@@ -54,7 +54,7 @@ def user(client: zitadel.Zitadel) -> Generator[UserServiceAddHumanUserResponse,
     yield response
     try:
         client.users.user_service_delete_user(response.user_id)  # type: ignore[arg-type]
-    except ApiException:
+    except ApiError:
         pass
 
 
@@ -113,5 +113,5 @@ def test_raises_api_exception_for_nonexistent_user(
         client: zitadel.Zitadel,
     ) -> None:
         """Raises an ApiException when retrieving a non-existent user."""
-        with pytest.raises(ApiException):
+        with pytest.raises(ApiError):
             client.users.user_service_get_user_by_id(str(uuid.uuid4()))

test/test_api_exception.py

@@ -0,0 +1,14 @@
+import unittest
+
+from zitadel_client.exceptions import ApiError
+
+
+class ApiExceptionTest(unittest.TestCase):
+    def test_api_exception(self) -> None:
+        headers = {"H": "v"}
+        e = ApiError(418, headers, "body")
+
+        self.assertEqual(str(e), "Error 418")
+        self.assertEqual(e.code, 418)
+        self.assertEqual(e.response_headers, headers)
+        self.assertEqual(e.response_body, "body")

zitadel_client/__init__.py

@@ -4,8 +4,8 @@
 from .api_response import ApiResponse  # noqa F401
 from .configuration import Configuration  # noqa F401
 from .exceptions import (
-    ApiException,  # noqa F401
-    OpenApiError,  # noqa F401
+    ApiError,  # noqa F401
+    ZitadelError,  # noqa F401
 )
 from .models import *  # noqa: F403, F401
 from .zitadel import Zitadel  # noqa F401

zitadel_client/api_client.py

@@ -22,7 +22,7 @@
 from zitadel_client.api_response import T as ApiResponseT
 from zitadel_client.auth.no_auth_authenticator import NoAuthAuthenticator
 from zitadel_client.configuration import Configuration
-from zitadel_client.exceptions import ApiException
+from zitadel_client.exceptions import ApiError
 
 RequestSerialized = Tuple[str, str, Dict[str, str], Optional[str], List[str]]
 
@@ -203,7 +203,7 @@ def call_api(
                 _request_timeout=_request_timeout,
             )
 
-        except ApiException as e:
+        except ApiError as e:
             raise e
 
         return response_data
@@ -246,11 +246,7 @@ def response_deserialize(
                 return_data = self.deserialize(response_text, response_type, content_type)
         finally:
             if not 200 <= response_data.status <= 299:
-                raise ApiException.from_response(
-                    http_resp=response_data,
-                    body=response_text,
-                    data=return_data,
-                )
+                raise ApiError(response_data.status, response_data.getheaders(), return_data)
             else:
                 return ApiResponse(
                     status_code=response_data.status,
@@ -339,7 +335,7 @@ def deserialize(self, response_text: str, response_type: str, content_type: Opti
         elif re.match(r"^text/[a-z.+-]+\s*(;|$)", content_type, re.IGNORECASE):
             data = response_text
         else:
-            raise ApiException(status=0, reason="Unsupported content type: {0}".format(content_type))
+            raise RuntimeError("Unsupported content type: {0}".format(content_type))
 
         return self.__deserialize(data, response_type)
 
@@ -600,7 +596,7 @@ def __deserialize_date(string):
         except ImportError:
             return string
         except ValueError as err:
-            raise rest.ApiException(status=0, reason="Failed to parse `{0}` as date object".format(string)) from err
+            raise RuntimeError("Failed to parse `{0}` as date object".format(string)) from err
 
     # noinspection PyNestedDecorators
     @no_type_check
@@ -618,7 +614,7 @@ def __deserialize_datetime(string):
         except ImportError:
             return string
         except ValueError as err:
-            raise rest.ApiException(status=0, reason=("Failed to parse `{0}` as datetime object".format(string))) from err
+            raise RuntimeError("Failed to parse `{0}` as datetime object".format(string)) from err
 
     # noinspection PyNestedDecorators
     @no_type_check
@@ -633,7 +629,7 @@ def __deserialize_enum(data, klass):
         try:
             return klass(data)
         except ValueError as err:
-            raise rest.ApiException(status=0, reason=("Failed to parse `{0}` as `{1}`".format(data, klass))) from err
+            raise RuntimeError("Failed to parse `{0}` as `{1}`".format(data, klass)) from err
 
     # noinspection PyNestedDecorators
     @no_type_check

zitadel_client/auth/oauth_authenticator.py

@@ -4,7 +4,7 @@
 
 from authlib.integrations.requests_client import OAuth2Session
 
-from zitadel_client import OpenApiError
+from zitadel_client import ZitadelError
 from zitadel_client.auth.authenticator import Authenticator, Token
 from zitadel_client.auth.open_id import OpenId
 
@@ -72,7 +72,7 @@ def refresh_token(self) -> Token:
             self.token = Token(access_token, expires_at)
             return self.token
         except Exception as e:
-            raise OpenApiError("Failed to refresh token: " + str(e)) from e
+            raise ZitadelError("Failed to refresh token: " + str(e)) from e
 
 
 T = TypeVar("T", bound="OAuthAuthenticatorBuilder[Any]")

zitadel_client/exceptions.py

@@ -1,81 +1,27 @@
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
-from typing_extensions import Self
 
-import zitadel_client.rest_response
+class ZitadelError(Exception):
+    """Base exception for all Zitadel API errors."""
 
 
-class OpenApiError(Exception):
-    """The base exception class for all OpenApiErrors"""
+class ApiError(ZitadelError):
+    """
+    Represents an HTTP error from the Zitadel API.
 
+    Attributes:
+      code             int HTTP status code
+      response_headers Dict[str, str] HTTP response headers
+      response_body Any HTTP response body
+    """
 
-class ApiException(OpenApiError):  # noqa N818 should be named with an Error suffix later
     def __init__(
         self,
-        status: Optional[int] = None,
-        reason: Optional[str] = None,
-        http_resp: Optional[zitadel_client.rest_response.RESTResponse] = None,
-        *,
-        body: Optional[str] = None,
-        data: Optional[Any] = None,
+        code: int,
+        response_headers: Dict[str, str],
+        response_body: Optional[Any],
     ) -> None:
-        self.status = status
-        self.reason = reason
-        self.body = body
-        self.data = data
-        self.headers = None
-
-        if http_resp:
-            if self.status is None:
-                self.status = http_resp.status
-            if self.reason is None:
-                self.reason = http_resp.reason
-            if self.body is None and http_resp.data is not None:
-                # noinspection PyBroadException
-                try:
-                    self.body = http_resp.data.decode("utf-8")
-                except Exception:  # noqa: S110
-                    pass
-            self.headers = http_resp.getheaders()
-
-    @classmethod
-    def from_response(
-        cls,
-        *,
-        http_resp: zitadel_client.rest_response.RESTResponse,
-        body: Optional[str],
-        data: Optional[Any],
-    ) -> Self:
-        if http_resp.status == 400:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        if http_resp.status == 401:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        if http_resp.status == 403:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        if http_resp.status == 404:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        # Added new conditions for 409 and 422
-        if http_resp.status == 409:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        if http_resp.status == 422:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-
-        if 500 <= http_resp.status <= 599:
-            raise ApiException(http_resp=http_resp, body=body, data=data)
-        raise ApiException(http_resp=http_resp, body=body, data=data)
-
-    def __str__(self) -> str:
-        """Custom error messages for exception"""
-        error_message = "({0})\nReason: {1}\n".format(self.status, self.reason)
-        if self.headers:
-            error_message += "HTTP response headers: {0}\n".format(self.headers)
-
-        if self.data or self.body:
-            error_message += "HTTP response body: {0}\n".format(self.data or self.body)
-
-        return error_message
+        super().__init__(f"Error {code}")
+        self.code = code
+        self.response_headers = response_headers
+        self.response_body = response_body