sandbox: Add error hierarchy

src/vercel/_internal/sandbox/__init__.py

@@ -1,10 +1,22 @@
 """Internal sandbox client implementations."""
 
 from vercel._internal.sandbox.core import AsyncSandboxOpsClient, SyncSandboxOpsClient
-from vercel._internal.sandbox.errors import APIError
+from vercel._internal.sandbox.errors import (
+    APIError,
+    SandboxAuthError,
+    SandboxError,
+    SandboxPermissionError,
+    SandboxRateLimitError,
+    SandboxServerError,
+)
 
 __all__ = [
     "AsyncSandboxOpsClient",
     "SyncSandboxOpsClient",
+    "SandboxError",
     "APIError",
+    "SandboxAuthError",
+    "SandboxPermissionError",
+    "SandboxRateLimitError",
+    "SandboxServerError",
 ]

src/vercel/_internal/sandbox/core.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import io
+import json
 import platform
 import posixpath
 import sys
@@ -27,7 +28,13 @@
     create_request_client,
 )
 from vercel._internal.iter_coroutine import iter_coroutine
-from vercel._internal.sandbox.errors import APIError
+from vercel._internal.sandbox.errors import (
+    APIError,
+    SandboxAuthError,
+    SandboxPermissionError,
+    SandboxRateLimitError,
+    SandboxServerError,
+)
 from vercel._internal.sandbox.models import (
     CommandFinishedResponse,
     CommandResponse,
@@ -63,8 +70,8 @@
 class SandboxRequestClient:
     """Low-level request layer wrapping a :class:`RequestClient`.
 
-    Translates non-2xx responses into :class:`APIError` and provides
-    a ``request_json`` convenience method.
+    Translates non-2xx responses into sandbox-specific :class:`APIError`
+    subclasses and provides a ``request_json`` convenience method.
     """
 
     def __init__(self, *, request_client: RequestClient) -> None:
@@ -96,36 +103,46 @@ async def request(
         if 200 <= response.status_code < 300:
             return response
 
+        error_body: bytes | None = None
+        try:
+            error_body = await response.aread()
+        except Exception:
+            try:
+                error_body = response.read()
+            except Exception:
+                error_body = None
+
         # Parse a helpful error message
         parsed: Any | None = None
         message = f"HTTP {response.status_code}"
-        try:
-            parsed = response.json()
-            if isinstance(parsed, dict):
-                if "message" in parsed and isinstance(parsed["message"], str):
-                    message = f"{message}: {parsed['message']}"
-                elif "error" in parsed:
-                    err = parsed["error"]
-                    if isinstance(err, dict):
-                        code = err.get("code")
-                        msg = err.get("message") or err.get("msg")
-                        if msg:
-                            message = f"{message}: {msg}"
-                        if code:
-                            message = f"{message} (code={code})"
-        except Exception:
-            parsed = None
+        if error_body:
+            try:
+                parsed = json.loads(error_body)
+                if isinstance(parsed, dict):
+                    if "message" in parsed and isinstance(parsed["message"], str):
+                        message = f"{message}: {parsed['message']}"
+                    elif "error" in parsed:
+                        err = parsed["error"]
+                        if isinstance(err, dict):
+                            code = err.get("code")
+                            msg = err.get("message") or err.get("msg")
+                            if msg:
+                                message = f"{message}: {msg}"
+                            if code:
+                                message = f"{message} (code={code})"
+            except Exception:
+                parsed = None
 
         if parsed is None:
             try:
-                text = response.text
+                text = error_body.decode() if error_body is not None else response.text
                 if text:
                     snippet = text if len(text) <= 500 else text[:500] + "\u2026"
                     message = f"{message}: {snippet}"
             except Exception:
                 pass
 
-        raise APIError(response, message, data=parsed)
+        raise _build_sandbox_error(response, message, data=parsed)
 
     async def request_json(
         self,
@@ -139,6 +156,29 @@ async def request_json(
         return r.json()
 
 
+def _build_sandbox_error(
+    response: httpx.Response,
+    message: str,
+    *,
+    data: Any | None = None,
+) -> APIError:
+    status_code = response.status_code
+    if status_code == 401:
+        return SandboxAuthError(response, message, data=data)
+    if status_code == 403:
+        return SandboxPermissionError(response, message, data=data)
+    if status_code == 429:
+        return SandboxRateLimitError(
+            response,
+            message,
+            data=data,
+            retry_after=response.headers.get("retry-after"),
+        )
+    if 500 <= status_code < 600:
+        return SandboxServerError(response, message, data=data)
+    return APIError(response, message, data=data)
+
+
 # ---------------------------------------------------------------------------
 # Tarball builder (pure Python, no I/O)
 # ---------------------------------------------------------------------------

src/vercel/_internal/sandbox/errors.py

@@ -7,12 +7,59 @@
 import httpx
 
 
-class APIError(Exception):
+def _normalize_retry_after(value: str | int | None) -> int | str | None:
+    if value is None:
+        return None
+    if isinstance(value, int):
+        return value
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return value
+
+
+class SandboxError(Exception):
+    """Base class for sandbox-specific errors."""
+
+
+class APIError(SandboxError):
     def __init__(self, response: httpx.Response, message: str, *, data: Any | None = None):
         super().__init__(message)
         self.response = response
         self.status_code = response.status_code
         self.data = data
 
 
-__all__ = ["APIError"]
+class SandboxAuthError(APIError):
+    """Authentication failures returned by the sandbox API."""
+
+
+class SandboxPermissionError(APIError):
+    """Authorization failures returned by the sandbox API."""
+
+
+class SandboxRateLimitError(APIError):
+    def __init__(
+        self,
+        response: httpx.Response,
+        message: str,
+        *,
+        data: Any | None = None,
+        retry_after: str | int | None = None,
+    ) -> None:
+        super().__init__(response, message, data=data)
+        self.retry_after: int | str | None = _normalize_retry_after(retry_after)
+
+
+class SandboxServerError(APIError):
+    """5xx responses returned by the sandbox API."""
+
+
+__all__ = [
+    "SandboxError",
+    "APIError",
+    "SandboxAuthError",
+    "SandboxPermissionError",
+    "SandboxRateLimitError",
+    "SandboxServerError",
+]

src/vercel/sandbox/__init__.py

@@ -1,4 +1,11 @@
-from vercel._internal.sandbox.errors import APIError
+from vercel._internal.sandbox import (
+    APIError,
+    SandboxAuthError,
+    SandboxError,
+    SandboxPermissionError,
+    SandboxRateLimitError,
+    SandboxServerError,
+)
 from vercel._internal.sandbox.network_policy import (
     NetworkPolicy,
     NetworkPolicyCustom,
@@ -14,7 +21,12 @@
 from .snapshot import AsyncSnapshot, Snapshot
 
 __all__ = [
+    "SandboxError",
     "APIError",
+    "SandboxAuthError",
+    "SandboxPermissionError",
+    "SandboxRateLimitError",
+    "SandboxServerError",
     "AsyncSandbox",
     "AsyncSandboxPage",
     "AsyncSnapshot",