feat: add error handling

src/codeocean/client.py

@@ -3,30 +3,62 @@
 from dataclasses import dataclass
 from requests_toolbelt.adapters.socket_options import TCPKeepAliveAdapter
 from requests_toolbelt.sessions import BaseUrlSession
-from typing import Optional
+from typing import Optional, Union
 from urllib3.util import Retry
+import requests
 
 from codeocean.capsule import Capsules
 from codeocean.computation import Computations
 from codeocean.data_asset import DataAssets
+from codeocean.errors import (
+    BadRequestError,
+    UnauthorizedError,
+    ForbiddenError,
+    NotFoundError,
+    InternalServerError,
+    CodeOceanError
+)
 
 
 @dataclass
 class CodeOcean:
 
     domain: str
     token: str
-    retries: Optional[Retry | int] = 0
+    retries: Optional[Union[Retry, int]] = 0
 
     def __post_init__(self):
         self.session = BaseUrlSession(base_url=f"{self.domain}/api/v1/")
         self.session.auth = (self.token, "")
         self.session.headers.update({"Content-Type": "application/json"})
-        self.session.hooks["response"] = [
-            lambda response, *args, **kwargs: response.raise_for_status()
-        ]
+        self.session.hooks["response"] = [self.error_handler]
         self.session.mount(self.domain, TCPKeepAliveAdapter(max_retries=self.retries))
 
         self.capsules = Capsules(client=self.session)
         self.computations = Computations(client=self.session)
         self.data_assets = DataAssets(client=self.session)
+
+    def error_handler(self, response, *args, **kwargs):
+        try:
+            response.raise_for_status()
+        except requests.HTTPError as ex:
+            try:
+                error_payload = response.json()
+                if not isinstance(error_payload, dict):
+                    raise ValueError("Response is not a JSON object")
+            except ValueError:
+                error_payload = {"message": response.text or str(ex)}
+
+            status_code = response.status_code
+            if status_code == 400:
+                raise BadRequestError.from_dict(error_payload).with_error(ex)
+            elif status_code == 401:
+                raise UnauthorizedError.from_dict(error_payload).with_error(ex)
+            elif status_code == 403:
+                raise ForbiddenError.from_dict(error_payload).with_error(ex)
+            elif status_code == 404:
+                raise NotFoundError.from_dict(error_payload).with_error(ex)
+            elif status_code >= 500:
+                raise InternalServerError.from_dict(error_payload).with_error(ex)
+            else:
+                raise CodeOceanError.from_dict(error_payload).with_error(ex)

src/codeocean/errors.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Optional, List, Type, TypeVar, Dict, Any
+import requests
+
+T = TypeVar("T", bound="Error")
+
+
+@dataclass
+class Error(Exception):
+    message: str
+    items: Optional[List[str]] = None
+    error: Optional[requests.HTTPError] = field(default=None, repr=False)
+
+    def with_error(self, ex: requests.HTTPError) -> Error:
+        self.error = ex
+        return self
+
+    @classmethod
+    def from_dict(cls: Type[T], payload: Dict[str, Any]) -> T:
+        message = payload.get("message", "An error occurred.")
+        items = payload.get("items") if isinstance(payload.get("items"), list) else None
+        return cls(message=message, items=items)
+
+    def __str__(self) -> str:
+        return self.message
+
+
+@dataclass
+class BadRequestError(Error):
+    """HTTP 400"""
+    pass
+
+
+@dataclass
+class UnauthorizedError(Error):
+    """HTTP 401"""
+    pass
+
+
+@dataclass
+class ForbiddenError(Error):
+    """HTTP 403"""
+    pass
+
+
+@dataclass
+class NotFoundError(Error):
+    """HTTP 404"""
+    pass
+
+
+@dataclass
+class InternalServerError(Error):
+    """HTTP 5xx"""
+    pass
+
+
+@dataclass
+class CodeOceanError(Error):
+    """Fallback for unexpected errors"""
+    pass