Add sandbox and snapshot list APIs

leap0/__init__.py

@@ -77,11 +77,13 @@
     from .models.sandbox import (
         CreateSandboxParams,
         NetworkPolicyMode,
+        SandboxListItem,
+        SandboxListResponse,
         SandboxState,
         SandboxStatus,
         sandbox_id_of,
     )
-    from .models.snapshot import CreateSnapshotParams, ResumeSnapshotParams, Snapshot, snapshot_id_of
+    from .models.snapshot import CreateSnapshotParams, ResumeSnapshotParams, Snapshot, SnapshotListResponse, snapshot_id_of
     from .models.ssh import SshAccess, SshValidation
     from .models.template import (
         AwsRegistryCredentials,
@@ -199,11 +201,14 @@
     "RegistryCredentialsInput",
     "ResumeSnapshotParams",
     "Sandbox",
+    "SandboxListItem",
+    "SandboxListResponse",
     "SandboxState",
     "SandboxStatus",
     "SandboxesClient",
     "SearchMatch",
     "Snapshot",
+    "SnapshotListResponse",
     "snapshot_id_of",
     "SnapshotsClient",
     "SshAccess",
@@ -262,12 +267,15 @@
     "Leap0WebSocketError": (".models.errors", "Leap0WebSocketError"),
     "CreateSandboxParams": (".models.sandbox", "CreateSandboxParams"),
     "NetworkPolicyMode": (".models.sandbox", "NetworkPolicyMode"),
+    "SandboxListItem": (".models.sandbox", "SandboxListItem"),
+    "SandboxListResponse": (".models.sandbox", "SandboxListResponse"),
     "SandboxStatus": (".models.sandbox", "SandboxStatus"),
     "SandboxState": (".models.sandbox", "SandboxState"),
     "sandbox_id_of": (".models.sandbox", "sandbox_id_of"),
     "CreateSnapshotParams": (".models.snapshot", "CreateSnapshotParams"),
     "ResumeSnapshotParams": (".models.snapshot", "ResumeSnapshotParams"),
     "Snapshot": (".models.snapshot", "Snapshot"),
+    "SnapshotListResponse": (".models.snapshot", "SnapshotListResponse"),
     "snapshot_id_of": (".models.snapshot", "snapshot_id_of"),
     "EditFileResult": (".models.filesystem", "EditFileResult"),
     "EditResult": (".models.filesystem", "EditResult"),

leap0/_async/sandbox.py

@@ -13,8 +13,15 @@
     DEFAULT_TIMEOUT_MIN,
     DEFAULT_VCPU,
 )
-from ..models.sandbox import CreateSandboxParams, Sandbox as SandboxData, SandboxRef, SandboxStatus, sandbox_id_of
-from .._schemas.sandbox import NetworkPolicyDict, SandboxCreateResponseDict, SandboxStatusResponseDict
+from ..models.sandbox import (
+    CreateSandboxParams,
+    Sandbox as SandboxData,
+    SandboxListResponse,
+    SandboxRef,
+    SandboxStatus,
+    sandbox_id_of,
+)
+from .._schemas.sandbox import ListSandboxesResponseDict, NetworkPolicyDict, SandboxCreateResponseDict, SandboxStatusResponseDict
 from .._utils.errors import intercept_errors
 from .._utils.url import ensure_leading_slash, sandbox_base_url, websocket_url_from_http
 from ._transport import AsyncTransport
@@ -147,6 +154,14 @@ def websocket_url(self, path: str = "/", *, port: int | None = None) -> str:
         """
         return self._client.sandboxes.websocket_url(self, path=path, port=port)
 
+    async def get_user_home_dir(self, http_timeout: float | None = None) -> str:
+        """Fetch the resolved home directory for the sandbox user."""
+        return await self._client.sandboxes.get_user_home_dir(self, http_timeout=http_timeout)
+
+    async def get_workdir(self, http_timeout: float | None = None) -> str:
+        """Fetch the configured working directory for the sandbox."""
+        return await self._client.sandboxes.get_workdir(self, http_timeout=http_timeout)
+
 
 def _inject_otel_env(env_vars: dict[str, str] | None) -> dict[str, str] | None:
     endpoint = os.environ.get(OTEL_EXPORTER_OTLP_ENDPOINT_ENV)
@@ -239,6 +254,59 @@ async def create(
         )
         return self._wrap_sandbox(SandboxData.from_dict(data))
 
+    @intercept_errors("Failed to list sandboxes: ")
+    async def list(
+        self,
+        *,
+        state: str | None = None,
+        sort: str = "created_at",
+        order_by: str = "desc",
+        page: int = 1,
+        page_size: int = 20,
+        http_timeout: float | None = None,
+    ) -> SandboxListResponse:
+        """List sandboxes for the authenticated organization.
+
+        Args:
+            state: Optional sandbox state filter.
+            sort: Sort field, either ``created_at`` or ``state``.
+            order_by: Sort direction, either ``asc`` or ``desc``.
+            page: 1-based page number.
+            page_size: Page size between 1 and 100.
+            http_timeout: Optional HTTP request timeout in seconds for this SDK call.
+
+        Returns:
+            SandboxListResponse: Paginated sandbox summaries.
+        """
+        valid_states = {"starting", "snapshotting", "running", "paused", "unpausing", "deleting"}
+        if state is not None and state not in valid_states:
+            raise ValueError(f"state must be one of {sorted(valid_states)}")
+        if sort not in {"created_at", "state"}:
+            raise ValueError("sort must be one of ['created_at', 'state']")
+        if order_by not in {"asc", "desc"}:
+            raise ValueError("order_by must be one of ['asc', 'desc']")
+        if page < 1:
+            raise ValueError("page must be at least 1")
+        if page_size < 1 or page_size > 100:
+            raise ValueError("page_size must be between 1 and 100")
+
+        params: dict[str, str | int] = {
+            "sort": sort,
+            "order-by": order_by,
+            "page": page,
+            "page-size": page_size,
+        }
+        if state is not None:
+            params["state"] = state
+
+        data = cast(ListSandboxesResponseDict, await self._transport.request_json(
+            "GET",
+            "/v1/sandboxes",
+            params=params,
+            timeout=http_timeout,
+        ))
+        return SandboxListResponse.from_dict(data)
+
     @intercept_errors("Failed to pause sandbox: ")
     async def pause(
         self,
@@ -293,6 +361,47 @@ async def delete(self, sandbox: SandboxRef, http_timeout: float | None = None) -
             timeout=http_timeout,
         )
 
+    @intercept_errors("Failed to get sandbox user home directory: ")
+    async def get_user_home_dir(self, sandbox: SandboxRef, http_timeout: float | None = None) -> str:
+        """Get the resolved home directory for the sandbox user.
+
+        Args:
+            sandbox: Sandbox ID or object.
+            http_timeout: Optional HTTP request timeout in seconds for this SDK call.
+
+        Returns:
+            str: Resolved sandbox user home directory.
+        """
+        data = cast(dict[str, object], await self._transport.request_json(
+            "GET", f"/v1/sandbox/{sandbox_id_of(sandbox)}/system/user-home-dir",
+            timeout=http_timeout,
+        ))
+        value = data.get("user_home_dir")
+        if not isinstance(value, str):
+            raise ValueError("Sandbox user home directory response missing 'user_home_dir'")
+        return value
+
+    @intercept_errors("Failed to get sandbox workdir: ")
+    async def get_workdir(self, sandbox: SandboxRef, http_timeout: float | None = None) -> str:
+        """Get the configured working directory for the sandbox.
+
+        Args:
+            sandbox: Sandbox ID or object.
+            http_timeout: Optional HTTP request timeout in seconds for this SDK call.
+
+        Returns:
+            str: Configured sandbox workdir.
+        """
+        data = cast(dict[str, object], await self._transport.request_json(
+            "GET", f"/v1/sandbox/{sandbox_id_of(sandbox)}/system/workdir",
+            timeout=http_timeout,
+        ))
+        value = data.get("workdir")
+        if not isinstance(value, str):
+            raise ValueError("Sandbox workdir response missing 'workdir'")
+        return value
+
+
     def invoke_url(self, sandbox: SandboxRef, path: str = "/", *, port: int | None = None) -> str:
         """Build an HTTPS URL for this sandbox.
 

leap0/_async/snapshots.py

@@ -4,8 +4,8 @@
 
 from .._internal.types import SandboxFactory
 from ..models.sandbox import Sandbox, SandboxRef, sandbox_id_of
-from ..models.snapshot import CreateSnapshotParams, ResumeSnapshotParams, Snapshot, SnapshotRef, snapshot_id_of
-from .._schemas.snapshot import SnapshotCreateResponseDict
+from ..models.snapshot import CreateSnapshotParams, ResumeSnapshotParams, Snapshot, SnapshotListResponse, SnapshotRef, snapshot_id_of
+from .._schemas.snapshot import ListSnapshotsResponseDict, SnapshotCreateResponseDict
 from .._schemas.sandbox import NetworkPolicyDict, SandboxCreateResponseDict
 from .._utils.errors import intercept_errors
 from ._transport import AsyncTransport
@@ -35,6 +35,43 @@ def _wrap_sandbox(self, sandbox: Sandbox) -> AsyncSnapshotSandboxT | Sandbox:
             return sandbox
         return self._sandbox_factory(sandbox)
 
+    @intercept_errors("Failed to list snapshots: ")
+    async def list(
+        self,
+        *,
+        query: str = "",
+        sort: str = "created_at",
+        order_by: str = "desc",
+        page: int = 1,
+        page_size: int = 20,
+        http_timeout: float | None = None,
+    ) -> SnapshotListResponse:
+        """List snapshots for the authenticated organization."""
+        if len(query) > 64:
+            raise ValueError("query must be at most 64 characters")
+        if sort not in {"created_at", "template_id"}:
+            raise ValueError("sort must be one of ['created_at', 'template_id']")
+        if order_by not in {"asc", "desc"}:
+            raise ValueError("order_by must be one of ['asc', 'desc']")
+        if page < 1:
+            raise ValueError("page must be at least 1")
+        if page_size < 1 or page_size > 100:
+            raise ValueError("page_size must be between 1 and 100")
+
+        data = cast(ListSnapshotsResponseDict, await self._transport.request_json(
+            "GET",
+            "/v1/snapshots",
+            params={
+                "query": query,
+                "sort": sort,
+                "order-by": order_by,
+                "page": page,
+                "page-size": page_size,
+            },
+            timeout=http_timeout,
+        ))
+        return SnapshotListResponse.from_dict(data)
+
     @intercept_errors("Failed to create snapshot: ")
     async def create(
         self,

leap0/_schemas/sandbox.py

@@ -41,3 +41,19 @@ class SandboxStatusResponseDict(TypedDict):
     state: SandboxState | str
     auto_pause: bool
     created_at: str
+
+class SandboxListItemResponseDict(TypedDict, total=False):
+    """Wire schema for sandbox list items."""
+    id: Required[str]
+    template_id: Required[str]
+    pod_id: Required[str]
+    state: Required[SandboxState | str]
+    launch_time: str
+    state_change_time: str
+    timeout_at: int
+    created_at: Required[str]
+
+class ListSandboxesResponseDict(TypedDict):
+    """Wire schema for paginated sandbox list responses."""
+    items: list[SandboxListItemResponseDict]
+    total_items: int

leap0/_schemas/snapshot.py

@@ -17,3 +17,8 @@ class SnapshotCreateResponseDict(TypedDict, total=False):
     state: SandboxState | str
     created_at: str
     network_policy: NetworkPolicyDict | None
+
+class ListSnapshotsResponseDict(TypedDict):
+    """Wire schema for paginated snapshot list responses."""
+    items: list[SnapshotCreateResponseDict]
+    total_items: int