diff --git a/src/blaxel/core/__init__.py b/src/blaxel/core/__init__.py index 219d1b23..e5a9d5b7 100644 --- a/src/blaxel/core/__init__.py +++ b/src/blaxel/core/__init__.py @@ -4,6 +4,7 @@ from .authentication import BlaxelAuth, auth, get_credentials from .cache import find_from_cache from .client.client import client +from .client.errors import BlaxelAPIError, error_to_exception from .common import ( autoload, env, @@ -78,4 +79,6 @@ "ImageInstance", "ImageBuildContext", "LocalFile", + "BlaxelAPIError", + "error_to_exception", ] diff --git a/src/blaxel/core/client/errors.py b/src/blaxel/core/client/errors.py index 5f92e76a..cda829b7 100644 --- a/src/blaxel/core/client/errors.py +++ b/src/blaxel/core/client/errors.py @@ -1,16 +1,64 @@ """Contains shared errors types that can be raised from API functions""" +from __future__ import annotations -class UnexpectedStatus(Exception): + +class BlaxelAPIError(Exception): + """Base exception for all Blaxel API errors. + + Allows users to catch all Blaxel API errors with a single except clause:: + + from blaxel.core import BlaxelAPIError + try: + ... + except BlaxelAPIError as e: + print(e.status_code, e.error_code) + """ + + def __init__( + self, + message: str, + status_code: int | None = None, + error_code: str | None = None, + response: object | None = None, + ): + super().__init__(message) + self.status_code = status_code + self.error_code = error_code + self.response = response + + +class UnexpectedStatus(BlaxelAPIError): """Raised by api functions when the response status an undocumented status and Client.raise_on_unexpected_status is True""" def __init__(self, status_code: int, content: bytes): - self.status_code = status_code self.content = content - super().__init__( - f"Unexpected status code: {status_code}\n\nResponse content:\n{content.decode(errors='ignore')}" + f"Unexpected status code: {status_code}\n\nResponse content:\n{content.decode(errors='ignore')}", + status_code=status_code, ) -__all__ = ["UnexpectedStatus"] +def error_to_exception(error: object) -> BlaxelAPIError: + """Convert an ``Error`` data-model instance to a ``BlaxelAPIError``. + + This is a free function rather than a method on ``Error`` because that class + is auto-generated and must not be edited directly. + """ + from .models.error import Error + from .types import UNSET + + if not isinstance(error, Error): + raise TypeError(f"Expected Error instance, got {type(error)}") + + status_code = error.code if error.code is not UNSET else None + message = error.message if error.message is not UNSET else error.error + + return BlaxelAPIError( + message=str(message), + status_code=status_code, + error_code=error.error, + ) + + +__all__ = ["BlaxelAPIError", "UnexpectedStatus", "error_to_exception"] diff --git a/src/blaxel/core/drive/drive.py b/src/blaxel/core/drive/drive.py index f9cb0f9f..723888c5 100644 --- a/src/blaxel/core/drive/drive.py +++ b/src/blaxel/core/drive/drive.py @@ -15,19 +15,18 @@ from ..client.api.drives.update_drive import asyncio as update_drive from ..client.api.drives.update_drive import sync as update_drive_sync from ..client.client import client -from ..client.errors import UnexpectedStatus +from ..client.errors import BlaxelAPIError, UnexpectedStatus from ..client.models import Drive, DriveSpec, Metadata from ..client.models.error import Error from ..client.types import UNSET from ..common.settings import settings -class DriveAPIError(Exception): +class DriveAPIError(BlaxelAPIError): """Exception raised when drive API returns an error.""" def __init__(self, message: str, status_code: int | None = None, code: str | None = None): - super().__init__(message) - self.status_code = status_code + super().__init__(message, status_code=status_code, error_code=code) self.code = code diff --git a/src/blaxel/core/sandbox/default/sandbox.py b/src/blaxel/core/sandbox/default/sandbox.py index 78ec28c9..5be064f6 100644 --- a/src/blaxel/core/sandbox/default/sandbox.py +++ b/src/blaxel/core/sandbox/default/sandbox.py @@ -12,6 +12,7 @@ from ...client.api.compute.list_sandboxes import asyncio as list_sandboxes from ...client.api.compute.update_sandbox import asyncio as update_sandbox from ...client.client import client +from ...client.errors import BlaxelAPIError from ...client.models import ( Metadata, MetadataLabels, @@ -44,12 +45,11 @@ from .system import SandboxSystem -class SandboxAPIError(Exception): +class SandboxAPIError(BlaxelAPIError): """Exception raised when sandbox API returns an error.""" def __init__(self, message: str, status_code: int | None = None, code: str | None = None): - super().__init__(message) - self.status_code = status_code + super().__init__(message, status_code=status_code, error_code=code) self.code = code diff --git a/src/blaxel/core/sandbox/types.py b/src/blaxel/core/sandbox/types.py index b43ed1fd..fb70c700 100644 --- a/src/blaxel/core/sandbox/types.py +++ b/src/blaxel/core/sandbox/types.py @@ -4,6 +4,7 @@ import httpx from attrs import define as _attrs_define +from ..client.errors import BlaxelAPIError from ..client.models import ( Env, Port, @@ -332,7 +333,7 @@ def __setattr__(self, name: str, value: Any) -> None: setattr(self._process_response, name, value) -class ResponseError(Exception): +class ResponseError(BlaxelAPIError): def __init__(self, response: httpx.Response): data_error = {} data = None @@ -348,8 +349,11 @@ def __init__(self, response: httpx.Response): if response.reason_phrase: data_error["statusText"] = response.reason_phrase - super().__init__(str(data_error)) - self.response = response + super().__init__( + str(data_error), + status_code=response.status_code, + response=response, + ) self.data = data self.error = None diff --git a/src/blaxel/core/volume/volume.py b/src/blaxel/core/volume/volume.py index 0ecfe5ca..fa3f4c6a 100644 --- a/src/blaxel/core/volume/volume.py +++ b/src/blaxel/core/volume/volume.py @@ -15,18 +15,18 @@ from ..client.api.volumes.update_volume import asyncio as update_volume from ..client.api.volumes.update_volume import sync as update_volume_sync from ..client.client import client +from ..client.errors import BlaxelAPIError from ..client.models import Metadata, Volume, VolumeSpec from ..client.models.error import Error from ..client.types import UNSET from ..common.settings import settings -class VolumeAPIError(Exception): +class VolumeAPIError(BlaxelAPIError): """Exception raised when volume API returns an error.""" def __init__(self, message: str, status_code: int | None = None, code: str | None = None): - super().__init__(message) - self.status_code = status_code + super().__init__(message, status_code=status_code, error_code=code) self.code = code