Beta submodule in Python SDK

Author: jakubno

packages/python-sdk/e2b/__init__.py

@@ -42,7 +42,7 @@
     NotEnoughSpaceException,
     TemplateException,
 )
-from .sandbox.sandbox_api import SandboxInfo, SandboxMetrics
+from .sandbox.sandbox_api import SandboxInfo, SandboxQuery, SandboxState, SandboxMetrics
 from .sandbox.commands.main import ProcessInfo
 from .sandbox.commands.command_handle import (
     CommandResult,
@@ -61,11 +61,13 @@
 from .sandbox_sync.main import Sandbox
 from .sandbox_sync.filesystem.watch_handle import WatchHandle
 from .sandbox_sync.commands.command_handle import CommandHandle
+from .sandbox_async.paginator import AsyncSandboxPaginator
 
 from .sandbox_async.utils import OutputHandler
 from .sandbox_async.main import AsyncSandbox
 from .sandbox_async.filesystem.watch_handle import AsyncWatchHandle
 from .sandbox_async.commands.command_handle import AsyncCommandHandle
+from .sandbox_sync.paginator import SandboxPaginator
 
 __all__ = [
     # API
@@ -86,6 +88,9 @@
     "SandboxInfo",
     "SandboxMetrics",
     "ProcessInfo",
+    "SandboxQuery",
+    "SandboxState",
+    "SandboxMetrics",
     # Command handle
     "CommandResult",
     "Stderr",
@@ -101,10 +106,12 @@
     "FileType",
     # Sync sandbox
     "Sandbox",
+    "SandboxPaginator",
     "WatchHandle",
     "CommandHandle",
     # Async sandbox
     "OutputHandler",
+    "AsyncSandboxPaginator",
     "AsyncSandbox",
     "AsyncWatchHandle",
     "AsyncCommandHandle",

packages/python-sdk/e2b/api/client/api/sandboxes/get_v2_sandboxes.py

@@ -1,8 +1,10 @@
 from dataclasses import dataclass
-from typing import Optional, Dict, Union
+from typing import Optional, Dict, Union, Unpack
 from datetime import datetime
 
+from e2b import ConnectionConfig
 from e2b.api.client.models import SandboxState, SandboxDetail, ListedSandbox
+from e2b.connection_config import ApiParams
 
 
 @dataclass
@@ -84,6 +86,9 @@ class SandboxQuery:
     metadata: Optional[dict[str, str]] = None
     """Filter sandboxes by metadata."""
 
+    state: Optional[list[SandboxState]] = None
+    """Filter sandboxes by state."""
+
 
 @dataclass
 class SandboxMetrics:
@@ -103,3 +108,34 @@ class SandboxMetrics:
     """Memory used in bytes."""
     timestamp: datetime
     """Timestamp of the metric entry."""
+
+
+class SandboxPaginatorBase:
+    def __init__(
+        self,
+        query: Optional[SandboxQuery] = None,
+        limit: Optional[int] = None,
+        next_token: Optional[str] = None,
+        **opts: Unpack[ApiParams],
+    ):
+        self._config = ConnectionConfig(**opts)
+
+        self.query = query
+        self.limit = limit
+
+        self._has_next = True
+        self._next_token = next_token
+
+    @property
+    def has_next(self) -> bool:
+        """
+        Returns True if there are more items to fetch.
+        """
+        return self._has_next
+
+    @property
+    def next_token(self) -> Optional[str]:
+        """
+        Returns the next token to use for pagination.
+        """
+        return self._next_token

packages/python-sdk/e2b/api/client/models/sandbox_detail.py

@@ -2,7 +2,14 @@
 import logging
 import httpx
 
-from typing import Dict, Optional, TypedDict, overload, List
+from typing import (
+    Dict,
+    Optional,
+    TypedDict,
+    overload,
+    List,
+    Type,
+)
 
 from packaging.version import Version
 from typing_extensions import Unpack
@@ -16,7 +23,7 @@
 from e2b.sandbox_async.filesystem.filesystem import Filesystem
 from e2b.sandbox_async.commands.command import Commands
 from e2b.sandbox_async.commands.pty import Pty
-from e2b.sandbox_async.sandbox_api import SandboxApi, SandboxInfo
+from e2b.sandbox_async.sandbox_api import SandboxApi, SandboxInfo, SandboxApiBeta
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +52,127 @@ class AsyncSandboxOpts(TypedDict):
     connection_config: ConnectionConfig
 
 
-class AsyncSandbox(SandboxApi):
+class _Beta(SandboxApiBeta):
+    def __init__(self, sandbox: "AsyncSandbox"):
+        self._sandbox = sandbox
+
+    @overload
+    async def pause(
+        self,
+        **opts: Unpack[ApiParams],
+    ) -> str:
+        """
+        Pause the sandbox.
+
+        :return: Sandbox ID that can be used to resume the sandbox
+        """
+        ...
+
+    @overload
+    @staticmethod
+    async def pause(
+        sandbox_id: str,
+        **opts: Unpack[ApiParams],
+    ) -> str:
+        """
+        Pause the sandbox specified by sandbox ID.
+
+        :param sandbox_id: Sandbox ID
+
+        :return: Sandbox ID that can be used to resume the sandbox
+        """
+        ...
+
+    @class_method_variant("_cls_pause")
+    async def pause(
+        self,
+        **opts: Unpack[ApiParams],
+    ) -> str:
+        """
+        Pause the sandbox.
+
+        :param request_timeout: Timeout for the request in **seconds**
+
+        :return: Sandbox ID that can be used to resume the sandbox
+        """
+
+        await self._cls_pause(
+            sandbox_id=self._sandbox.sandbox_id,
+            **opts,
+        )
+
+        return self._sandbox.sandbox_id
+
+    @overload
+    async def resume(
+        self,
+        timeout: Optional[int] = None,
+        **opts: Unpack[ApiParams],
+    ) -> "AsyncSandbox":
+        """
+        Resume the sandbox.
+
+        :return: A running sandbox instance
+        """
+        ...
+
+    @overload
+    @staticmethod
+    async def resume(
+        sandbox_id: str,
+        timeout: Optional[int] = None,
+        **opts: Unpack[ApiParams],
+    ) -> "AsyncSandbox":
+        """
+        Resume the sandbox.
+
+        :param sandbox_id: Sandbox ID
+        :param timeout: Timeout for the sandbox in **seconds**
+
+        :return: A running sandbox instance
+        """
+        ...
+
+    @class_method_variant("_cls_resume")
+    async def resume(
+        self,
+        timeout: Optional[int] = None,
+        **opts: Unpack[ApiParams],
+    ) -> "AsyncSandbox":
+        """
+        Resume the sandbox.
+
+        The **default sandbox timeout of 300 seconds** will be used for the resumed sandbox.
+        If you pass a custom timeout via the `timeout` parameter, it will be used instead.
+
+        :param timeout: Timeout for the sandbox in **seconds**
+
+        :return: A running sandbox instance
+        """
+
+        print(timeout)
+        await self._cls_resume(
+            sandbox_id=self._sandbox.sandbox_id,
+            timeout=timeout,
+            **opts,
+        )
+
+        return await self._sandbox.connect(
+            sandbox_id=self._sandbox.sandbox_id,
+            **opts,
+        )
+
+
+class _AsyncSandboxMeta(type):
+    """Metaclass for AsyncSandbox to provide class-level beta access."""
+
+    @property
+    def beta(cls) -> Type[_Beta]:
+        """Access to beta features at class level."""
+        return _Beta
+
+
+class AsyncSandbox(SandboxApi, metaclass=_AsyncSandboxMeta):
     """
     E2B cloud sandbox is a secure and isolated cloud environment.
 
@@ -89,6 +216,13 @@ def pty(self) -> Pty:
         """
         return self._pty
 
+    @property
+    def beta(self) -> _Beta:
+        """
+        Module for beta features.
+        """
+        return self._beta
+
     def __init__(self, **opts: Unpack[AsyncSandboxOpts]):
         """
         Use `AsyncSandbox.create()` to create a new sandbox instead.
@@ -120,6 +254,7 @@ def __init__(self, **opts: Unpack[AsyncSandboxOpts]):
             self.connection_config,
             self._transport.pool,
         )
+        self._beta = _Beta(self)
 
     async def is_running(self, request_timeout: Optional[float] = None) -> bool:
         """
@@ -265,7 +400,7 @@ async def connect(
             sandbox_id=sandbox_id,
             sandbox_domain=response.sandbox_domain,
             connection_config=connection_config,
-            envd_version=response.envd_version,
+            envd_version=response._envd_version,
             envd_access_token=envd_access_token,
         )
 

packages/python-sdk/e2b/sandbox/sandbox_api.py

@@ -0,0 +1,72 @@
+import urllib.parse
+from typing import Optional, List
+
+from e2b.api.client.api.sandboxes import get_v2_sandboxes
+from e2b.api.client.types import UNSET
+from e2b.exceptions import SandboxException
+from e2b.sandbox.main import SandboxBase
+from e2b.sandbox.sandbox_api import SandboxPaginatorBase, SandboxInfo
+from e2b.api import AsyncApiClient, handle_api_exception
+from e2b.api.client.models.error import Error
+
+
+class AsyncSandboxPaginator(SandboxPaginatorBase):
+    """
+    Paginator for listing sandboxes.
+
+    Example:
+    ```python
+    paginator = AsyncSandbox.list()
+
+    while paginator.has_next:
+        sandboxes = await paginator.next_items()
+        print(sandboxes)
+    ```
+    """
+
+    async def next_items(self) -> List[SandboxInfo]:
+        """
+        Returns the next page of sandboxes.
+
+        Call this method only if `has_next` is `True`, otherwise it will raise an exception.
+
+        :returns: List of sandboxes
+        """
+        if not self.has_next:
+            raise Exception("No more items to fetch")
+
+        # Convert filters to the format expected by the API
+        metadata: Optional[str] = None
+        if self.query and self.query.metadata:
+            quoted_metadata = {
+                urllib.parse.quote(k): urllib.parse.quote(v)
+                for k, v in self.query.metadata.items()
+            }
+            metadata = urllib.parse.urlencode(quoted_metadata)
+
+        async with AsyncApiClient(
+            self._config,
+            limits=SandboxBase._limits,
+        ) as api_client:
+            res = await get_v2_sandboxes.asyncio_detailed(
+                client=api_client,
+                metadata=metadata if metadata else UNSET,
+                state=self.query.state if self.query and self.query.state else UNSET,
+                limit=self.limit if self.limit else UNSET,
+                next_token=self._next_token if self._next_token else UNSET,
+            )
+
+            if res.status_code >= 300:
+                raise handle_api_exception(res)
+
+            self._next_token = res.headers.get("x-next-token")
+            self._has_next = bool(self._next_token)
+
+            if res.parsed is None:
+                return []
+
+            # Check if res.parse is Error
+            if isinstance(res.parsed, Error):
+                raise SandboxException(f"{res.parsed.message}: Request failed")
+
+            return [SandboxInfo._from_listed_sandbox(sandbox) for sandbox in res.parsed]