feat(excetions): add base exceptions (#39)

Datata1 · web-flow · commit 8ad8739c2b21 · 2026-02-07T14:24:35.000+01:00
Co-authored-by: Datata1 &lt;&gt;
diff --git a/src/codesphere/__init__.py b/src/codesphere/__init__.py
@@ -20,34 +20,55 @@
 """
 
 import logging
-from .client import CodesphereSDK
 
-from .exceptions import CodesphereError, AuthenticationError
+from .client import CodesphereSDK
+from .exceptions import (
+    APIError,
+    AuthenticationError,
+    AuthorizationError,
+    CodesphereError,
+    ConflictError,
+    NetworkError,
+    NotFoundError,
+    RateLimitError,
+    TimeoutError,
+    ValidationError,
+)
+from .resources.metadata import Characteristic, Datacenter, Image, WsPlan
 from .resources.team import (
-    Team,
-    TeamCreate,
-    TeamBase,
-    Domain,
     CustomDomainConfig,
-    DomainVerificationStatus,
+    Domain,
     DomainBase,
     DomainRouting,
+    DomainVerificationStatus,
+    Team,
+    TeamBase,
+    TeamCreate,
 )
 from .resources.workspace import (
     Workspace,
     WorkspaceCreate,
-    WorkspaceUpdate,
     WorkspaceStatus,
+    WorkspaceUpdate,
 )
 from .resources.workspace.envVars import EnvVar
-from .resources.metadata import Datacenter, Characteristic, WsPlan, Image
 
 logging.getLogger("codesphere").addHandler(logging.NullHandler())
 
 __all__ = [
     "CodesphereSDK",
+    # Exceptions
     "CodesphereError",
     "AuthenticationError",
+    "AuthorizationError",
+    "NotFoundError",
+    "ValidationError",
+    "ConflictError",
+    "RateLimitError",
+    "APIError",
+    "NetworkError",
+    "TimeoutError",
+    # Resources
     "Team",
     "TeamCreate",
     "TeamBase",
@@ -64,6 +85,5 @@
     "CustomDomainConfig",
     "DomainVerificationStatus",
     "DomainBase",
-    "DomainsResource",
     "DomainRouting",
 ]
diff --git a/src/codesphere/exceptions.py b/src/codesphere/exceptions.py
@@ -1,16 +1,230 @@
+from typing import Any, Optional
+
+import httpx
+
+
 class CodesphereError(Exception):
-    """Base exception class for all errors in the Codesphere SDK."""
+    """Base exception class for all errors in the Codesphere SDK.
+
+    All SDK exceptions inherit from this, so users can catch this
+    to handle any SDK-related error.
+    """
 
-    pass
+    def __init__(self, message: str = "An error occurred in the Codesphere SDK."):
+        self.message = message
+        super().__init__(self.message)
 
 
 class AuthenticationError(CodesphereError):
-    """Raised for authentication-related errors, like a missing API token."""
+    """Raised for authentication-related errors, like a missing or invalid API token.
 
-    def __init__(self, message: str = None):
+    HTTP Status: 401
+    """
+
+    def __init__(self, message: Optional[str] = None):
         if message is None:
             message = (
-                "Authentication token not provided. Please pass it as an argument "
+                "Authentication token not provided or invalid. Please pass it as an argument "
                 "or set the 'CS_TOKEN' environment variable."
             )
         super().__init__(message)
+
+
+class AuthorizationError(CodesphereError):
+    """Raised when the user doesn't have permission to perform an action.
+
+    HTTP Status: 403
+    """
+
+    def __init__(self, message: Optional[str] = None):
+        if message is None:
+            message = "You don't have permission to perform this action."
+        super().__init__(message)
+
+
+class NotFoundError(CodesphereError):
+    """Raised when the requested resource does not exist.
+
+    HTTP Status: 404
+    """
+
+    def __init__(self, message: Optional[str] = None, resource: Optional[str] = None):
+        self.resource = resource
+        if message is None:
+            if resource:
+                message = f"The requested {resource} was not found."
+            else:
+                message = "The requested resource was not found."
+        super().__init__(message)
+
+
+class ValidationError(CodesphereError):
+    """Raised when the request data is invalid or malformed.
+
+    HTTP Status: 400, 422
+    """
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        errors: Optional[list[dict[str, Any]]] = None,
+    ):
+        self.errors = errors or []
+        if message is None:
+            message = "The request data was invalid."
+        super().__init__(message)
+
+
+class ConflictError(CodesphereError):
+    """Raised when there's a conflict with the current state of a resource.
+
+    HTTP Status: 409
+    """
+
+    def __init__(self, message: Optional[str] = None):
+        if message is None:
+            message = "The request conflicts with the current state of the resource."
+        super().__init__(message)
+
+
+class RateLimitError(CodesphereError):
+    """Raised when rate limits are exceeded.
+
+    HTTP Status: 429
+    """
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        retry_after: Optional[int] = None,
+    ):
+        self.retry_after = retry_after
+        if message is None:
+            if retry_after:
+                message = f"Rate limit exceeded. Retry after {retry_after} seconds."
+            else:
+                message = "Rate limit exceeded. Please slow down your requests."
+        super().__init__(message)
+
+
+class APIError(CodesphereError):
+    """Raised for general API errors that don't fit other categories.
+
+    Contains detailed information about the failed request.
+    """
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        status_code: Optional[int] = None,
+        response_body: Optional[Any] = None,
+        request_url: Optional[str] = None,
+        request_method: Optional[str] = None,
+    ):
+        self.status_code = status_code
+        self.response_body = response_body
+        self.request_url = request_url
+        self.request_method = request_method
+
+        if message is None:
+            message = f"API request failed with status {status_code}."
+        super().__init__(message)
+
+    def __str__(self) -> str:
+        parts = [self.message]
+        if self.status_code:
+            parts.append(f"Status: {self.status_code}")
+        if self.request_method and self.request_url:
+            parts.append(f"Request: {self.request_method} {self.request_url}")
+        return " | ".join(parts)
+
+
+class NetworkError(CodesphereError):
+    """Raised for network-related issues like connection failures or timeouts."""
+
+    def __init__(
+        self, message: Optional[str] = None, original_error: Optional[Exception] = None
+    ):
+        self.original_error = original_error
+        if message is None:
+            message = "A network error occurred while connecting to the API."
+        super().__init__(message)
+
+
+class TimeoutError(NetworkError):
+    """Raised when a request times out."""
+
+    def __init__(self, message: Optional[str] = None):
+        if message is None:
+            message = "The request timed out. The server may be slow or unavailable."
+        super().__init__(message)
+
+
+def raise_for_status(response: httpx.Response) -> None:
+    """Convert HTTP errors to appropriate SDK exceptions.
+
+    This function should be called after every API request to translate
+    HTTP errors into user-friendly SDK exceptions.
+
+    Args:
+        response: The httpx Response object to check.
+
+    Raises:
+        AuthenticationError: For 401 responses.
+        AuthorizationError: For 403 responses.
+        NotFoundError: For 404 responses.
+        ValidationError: For 400/422 responses.
+        ConflictError: For 409 responses.
+        RateLimitError: For 429 responses.
+        APIError: For other 4xx/5xx responses.
+    """
+    if response.is_success:
+        return
+
+    status_code = response.status_code
+
+    error_message = None
+    response_body = None
+    try:
+        response_body = response.json()
+        error_message = (
+            response_body.get("message")
+            or response_body.get("error")
+            or response_body.get("detail")
+            or response_body.get("errors")
+        )
+        if isinstance(error_message, list):
+            error_message = "; ".join(str(e) for e in error_message)
+    except Exception:
+        error_message = response.text or None
+
+    request_url = str(response.request.url) if response.request else None
+    request_method = response.request.method if response.request else None
+
+    if status_code == 401:
+        raise AuthenticationError(error_message)
+    elif status_code == 403:
+        raise AuthorizationError(error_message)
+    elif status_code == 404:
+        raise NotFoundError(error_message)
+    elif status_code in (400, 422):
+        errors = (
+            response_body.get("errors") if isinstance(response_body, dict) else None
+        )
+        raise ValidationError(error_message, errors=errors)
+    elif status_code == 409:
+        raise ConflictError(error_message)
+    elif status_code == 429:
+        retry_after = response.headers.get("Retry-After")
+        retry_after_int = (
+            int(retry_after) if retry_after and retry_after.isdigit() else None
+        )
+        raise RateLimitError(error_message, retry_after=retry_after_int)
+    else:
+        raise APIError(
+            message=error_message,
+            status_code=status_code,
+            response_body=response_body,
+            request_url=request_url,
+            request_method=request_method,
+        )
diff --git a/src/codesphere/http_client.py b/src/codesphere/http_client.py
@@ -1,9 +1,12 @@
-from functools import partial
 import logging
+from functools import partial
+from typing import Any, Optional
+
 import httpx
 from pydantic import BaseModel
-from typing import Optional, Any
+
 from .config import settings
+from .exceptions import NetworkError, TimeoutError, raise_for_status
 
 log = logging.getLogger(__name__)
 
@@ -68,18 +71,21 @@ async def request(
                 f"Response: {response.status_code} {response.reason_phrase} for {method} {endpoint}"
             )
 
-            response.raise_for_status()
+            raise_for_status(response)
             return response
 
-        except httpx.HTTPStatusError as e:
-            log.error(
-                f"HTTP Error {e.response.status_code} for {e.request.method} {e.request.url}"
-            )
-            try:
-                log.error(f"Error Response Body: {e.response.json()}")
-            except Exception:
-                log.error(f"Error Response Body (non-json): {e.response.text}")
-            raise e
-        except Exception as e:
-            log.error(f"An unexpected error occurred: {e}")
-            raise e
+        except httpx.TimeoutException as e:
+            log.error(f"Request timeout for {method} {endpoint}: {e}")
+            raise TimeoutError(f"Request to {endpoint} timed out.") from e
+        except httpx.ConnectError as e:
+            log.error(f"Connection error for {method} {endpoint}: {e}")
+            raise NetworkError(
+                f"Failed to connect to the API: {e}",
+                original_error=e,
+            ) from e
+        except httpx.RequestError as e:
+            log.error(f"Network error for {method} {endpoint}: {e}")
+            raise NetworkError(
+                f"A network error occurred: {e}",
+                original_error=e,
+            ) from e
diff --git a/src/codesphere/resources/workspace/resources.py b/src/codesphere/resources/workspace/resources.py
@@ -1,15 +1,16 @@
 from typing import List
+
 from pydantic import Field
 
+from ...core import ResourceBase
 from ...core.base import ResourceList
 from ...core.operations import AsyncCallable
+from ...exceptions import ValidationError
 from .operations import (
     _CREATE_OP,
     _GET_OP,
     _LIST_BY_TEAM_OP,
 )
-
-from ...core import ResourceBase
 from .schemas import Workspace, WorkspaceCreate
 
 
@@ -19,12 +20,16 @@ class WorkspacesResource(ResourceBase):
     )
 
     async def list(self, team_id: int) -> List[Workspace]:
+        if team_id <= 0:
+            raise ValidationError("team_id must be a positive integer")
         result = await self.list_by_team_op(team_id=team_id)
         return result.root
 
     get_op: AsyncCallable[Workspace] = Field(default=_GET_OP, exclude=True)
 
     async def get(self, workspace_id: int) -> Workspace:
+        if workspace_id <= 0:
+            raise ValidationError("workspace_id must be a positive integer")
         return await self.get_op(workspace_id=workspace_id)
 
     create_op: AsyncCallable[Workspace] = Field(default=_CREATE_OP, exclude=True)
diff --git a/tests/conftest.py b/tests/conftest.py
diff --git a/tests/test_client.py b/tests/test_client.py
