feat: introduce SeamHttpUnauthorizedError and SeamHttpInvalidInputError (#99)

* Improve http error handling, add invalid input and unauthorized seam errors

* Export SeamHttpUnauthorizedError and SeamHttpInvalidInputError

* Test new error handling

* ci: Format code

* Lint fixes

* Remove 200 status check from _handle_error_response

---------

Co-authored-by: Seam Bot <devops@getseam.com>
Co-authored-by: Evan Sosenko <evan@getseam.com>

Author: 3 people

seam/__init__.py

@@ -7,6 +7,8 @@
 from .auth import SeamInvalidTokenError
 from .exceptions import (
     SeamHttpApiError,
+    SeamHttpUnauthorizedError,
+    SeamHttpInvalidInputError,
     SeamActionAttemptError,
     SeamActionAttemptFailedError,
     SeamActionAttemptTimeoutError,

seam/client.py

@@ -4,7 +4,11 @@
 from importlib.metadata import version
 
 from .constants import LTS_VERSION
-from .exceptions import SeamHttpApiError
+from .exceptions import (
+    SeamHttpApiError,
+    SeamHttpInvalidInputError,
+    SeamHttpUnauthorizedError,
+)
 from .models import AbstractSeamHttpClient
 
 SDK_HEADERS = {
@@ -30,10 +34,33 @@ def request(self, method, url, *args, **kwargs):
         return self._handle_response(response)
 
     def _handle_response(self, response: requests.Response):
-        if response.status_code != 200:
-            raise SeamHttpApiError(response)
+        if not 200 <= response.status_code < 300:
+            self._handle_error_response(response)
 
-        if "application/json" in response.headers["content-type"]:
+        if "application/json" in response.headers.get("content-type", ""):
             return response.json()
 
         return response.text
+
+    def _handle_error_response(self, response: requests.Response):
+        status_code = response.status_code
+        request_id = response.headers.get("seam-request-id")
+
+        if status_code == 401:
+            raise SeamHttpUnauthorizedError(request_id)
+
+        error = response.json().get("error", {})
+        error_type = error.get("type", "unknown_error")
+        error_message = error.get("message", "Unknown error")
+        error_data = error.get("data", None)
+
+        error_details = {
+            "type": error_type,
+            "message": error_message,
+            "data": error_data,
+        }
+
+        if error_type == "invalid_input":
+            raise SeamHttpInvalidInputError(error_details, status_code, request_id)
+
+        raise SeamHttpApiError(error_details, status_code, request_id)

seam/exceptions.py

@@ -1,26 +1,31 @@
+from typing import Any, Dict
 from niquests import Response
 from .routes.models import ActionAttempt
 
 
 # HTTP
 class SeamHttpApiError(Exception):
-    def __init__(
-        self,
-        response: Response,
-    ):
-        self.status_code = response.status_code
-        self.request_id = response.headers.get("seam-request-id", None)
-
-        self.metadata = None
-        if "application/json" in response.headers["content-type"]:
-            parsed_response = response.json()
-            self.metadata = parsed_response.get("error", None)
+    def __init__(self, error: Dict[str, Any], status_code: int, request_id: str):
+        super().__init__(error["message"])
+        self.code = error["type"]
+        self.status_code = status_code
+        self.request_id = request_id
+        self.data = error.get("data")
 
+
+class SeamHttpUnauthorizedError(SeamHttpApiError):
+    def __init__(self, request_id: str):
         super().__init__(
-            f"SeamApiException: status={self.status_code}, request_id={self.request_id}, metadata={self.metadata}"
+            {"type": "unauthorized", "message": "Unauthorized"}, 401, request_id
         )
 
 
+class SeamHttpInvalidInputError(SeamHttpApiError):
+    def __init__(self, error: Dict[str, Any], status_code: int, request_id: str):
+        super().__init__(error, status_code, request_id)
+        self.code = "invalid_input"
+
+
 # Action Attempt
 class SeamActionAttemptError(Exception):
     def __init__(self, message: str, action_attempt: ActionAttempt):

seam/models.py

@@ -19,6 +19,10 @@ def request(self, method: str, url: str, *args, **kwargs):
     def _handle_response(self, response: requests.Response):
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def _handle_error_response(self, response: requests.Response):
+        raise NotImplementedError
+
 
 class AbstractSeam(AbstractRoutes):
     lts_version: str

seam/seam.py

@@ -1,5 +1,4 @@
 from typing import Any, Optional, Union, Dict
-import niquests as requests
 from typing_extensions import Self
 
 from .constants import LTS_VERSION

test/test_client.py

@@ -0,0 +1,46 @@
+import pytest
+from seam import Seam
+from seam.exceptions import (
+    SeamHttpApiError,
+    SeamHttpInvalidInputError,
+    SeamHttpUnauthorizedError,
+)
+
+
+def test_seam_http_throws_unauthorized_error(server):
+    endpoint, _ = server
+
+    seam = Seam(api_key="seam_invalid_api_key", endpoint=endpoint)
+
+    with pytest.raises(SeamHttpUnauthorizedError) as exc_info:
+        seam.devices.list()
+    err = exc_info.value
+    assert err.status_code == 401
+    assert err.code == "unauthorized"
+    assert err.request_id.startswith("request")
+
+
+def test_seam_http_throws_api_error_on_standard_error_response(server):
+    endpoint, seed = server
+
+    seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint)
+
+    with pytest.raises(SeamHttpApiError) as exc_info:
+        seam.devices.get(device_id="unknown-device")
+    err = exc_info.value
+    assert err.status_code == 404
+    assert err.code == "device_not_found"
+    assert err.request_id.startswith("request")
+
+
+def test_seam_http_throws_invalid_input_error(server):
+    endpoint, seed = server
+
+    seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint)
+
+    with pytest.raises(SeamHttpInvalidInputError) as exc_info:
+        seam.devices.get(device_id=4242)
+    err = exc_info.value
+    assert err.status_code == 400
+    assert err.code == "invalid_input"
+    assert err.request_id.startswith("request")