feat: add new storage error class

storage3/_async/bucket.py

@@ -2,10 +2,11 @@
 
 from typing import Any, Optional
 
-from httpx import HTTPError, Response
+from httpx import HTTPStatusError, Response
 
+from ..exceptions import StorageApiError
 from ..types import CreateOrUpdateBucketOptions, RequestMethod
-from ..utils import AsyncClient, StorageException
+from ..utils import AsyncClient
 from .file_api import AsyncBucket
 
 __all__ = ["AsyncStorageBucketAPI"]
@@ -23,13 +24,12 @@ async def _request(
         url: str,
         json: Optional[dict[Any, Any]] = None,
     ) -> Response:
-        response = await self._client.request(method, url, json=json)
         try:
+            response = await self._client.request(method, url, json=json)
             response.raise_for_status()
-        except HTTPError:
-            raise StorageException(
-                {**response.json(), "statusCode": response.status_code}
-            )
+        except HTTPStatusError as exc:
+            resp = exc.response.json()
+            raise StorageApiError(resp["message"], resp["error"], resp["statusCode"])
 
         return response
 

storage3/_async/file_api.py

@@ -3,13 +3,13 @@
 import urllib.parse
 from dataclasses import dataclass, field
 from io import BufferedReader, FileIO
-from json import JSONDecodeError
 from pathlib import Path
 from typing import Any, Literal, Optional, Union, cast
 
-from httpx import HTTPError, Response
+from httpx import HTTPStatusError, Response
 
 from ..constants import DEFAULT_FILE_OPTIONS, DEFAULT_SEARCH_OPTIONS
+from ..exceptions import StorageApiError
 from ..types import (
     BaseBucket,
     CreateSignedURLsOptions,
@@ -47,12 +47,9 @@ async def _request(
                 method, url, headers=headers or {}, json=json, files=files, **kwargs
             )
             response.raise_for_status()
-        except HTTPError:
-            try:
-                resp = response.json()
-                raise StorageException({**resp, "statusCode": response.status_code})
-            except JSONDecodeError:
-                raise StorageException({"statusCode": response.status_code})
+        except HTTPStatusError as exc:
+            resp = exc.response.json()
+            raise StorageApiError(resp["message"], resp["error"], resp["statusCode"])
 
         return response
 

storage3/_sync/bucket.py

@@ -2,10 +2,11 @@
 
 from typing import Any, Optional
 
-from httpx import HTTPError, Response
+from httpx import HTTPStatusError, Response
 
+from ..exceptions import StorageApiError
 from ..types import CreateOrUpdateBucketOptions, RequestMethod
-from ..utils import StorageException, SyncClient
+from ..utils import SyncClient
 from .file_api import SyncBucket
 
 __all__ = ["SyncStorageBucketAPI"]
@@ -23,13 +24,12 @@ def _request(
         url: str,
         json: Optional[dict[Any, Any]] = None,
     ) -> Response:
-        response = self._client.request(method, url, json=json)
         try:
+            response = self._client.request(method, url, json=json)
             response.raise_for_status()
-        except HTTPError:
-            raise StorageException(
-                {**response.json(), "statusCode": response.status_code}
-            )
+        except HTTPStatusError as exc:
+            resp = exc.response.json()
+            raise StorageApiError(resp["message"], resp["error"], resp["statusCode"])
 
         return response
 

storage3/_sync/file_api.py

@@ -3,13 +3,13 @@
 import urllib.parse
 from dataclasses import dataclass, field
 from io import BufferedReader, FileIO
-from json import JSONDecodeError
 from pathlib import Path
 from typing import Any, Literal, Optional, Union, cast
 
-from httpx import HTTPError, Response
+from httpx import HTTPStatusError, Response
 
 from ..constants import DEFAULT_FILE_OPTIONS, DEFAULT_SEARCH_OPTIONS
+from ..exceptions import StorageApiError
 from ..types import (
     BaseBucket,
     CreateSignedURLsOptions,
@@ -47,12 +47,9 @@ def _request(
                 method, url, headers=headers or {}, json=json, files=files, **kwargs
             )
             response.raise_for_status()
-        except HTTPError:
-            try:
-                resp = response.json()
-                raise StorageException({**resp, "statusCode": response.status_code})
-            except JSONDecodeError:
-                raise StorageException({"statusCode": response.status_code})
+        except HTTPStatusError as exc:
+            resp = exc.response.json()
+            raise StorageApiError(resp["message"], resp["error"], resp["statusCode"])
 
         return response
 

storage3/exceptions.py

@@ -0,0 +1,33 @@
+from typing import TypedDict
+
+from .utils import StorageException
+
+
+class StorageApiErrorDict(TypedDict):
+    name: str
+    message: str
+    status: int
+
+
+class StorageApiError(StorageException):
+    """Error raised when an operation on the storage API fails."""
+
+    def __init__(self, message: str, code: str, status: int) -> None:
+        error_message = "{{'statusCode': {}, 'error': {}, 'message': {}}}".format(
+            status,
+            code,
+            message,
+        )
+        super().__init__(error_message)
+        self.name = "StorageApiError"
+        self.message = message
+        self.code = code
+        self.status = status
+
+    def to_dict(self) -> StorageApiErrorDict:
+        return {
+            "name": self.name,
+            "code": self.code,
+            "message": self.message,
+            "status": self.status,
+        }