modal.sandbox

modal.sandbox.Sandbox

class Sandbox(modal.object.Object)

A Sandbox object lets you interact with a running sandbox. This API is similar to Python’s asyncio.subprocess.Process.

Refer to the guide on how to spawn and use sandboxes.

def __init__(self, *args, **kwargs):

create

@staticmethod
def create(
    *entrypoint_args: str,
    app: Optional["modal.app._App"] = None,  # Optionally associate the sandbox with an app
    environment_name: Optional[str] = None,  # Optionally override the default environment
    image: Optional[_Image] = None,  # The image to run as the container for the sandbox.
    mounts: Sequence[_Mount] = (),  # Mounts to attach to the sandbox.
    secrets: Sequence[_Secret] = (),  # Environment variables to inject into the sandbox.
    network_file_systems: Dict[Union[str, os.PathLike], _NetworkFileSystem] = {},
    timeout: Optional[int] = None,  # Maximum execution time of the sandbox in seconds.
    workdir: Optional[str] = None,  # Working directory of the sandbox.
    gpu: GPU_T = None,
    cloud: Optional[str] = None,
    region: Optional[Union[str, Sequence[str]]] = None,  # Region or regions to run the sandbox on.
    cpu: Optional[float] = None,  # How many CPU cores to request. This is a soft limit.
    # Specify, in MiB, a memory request which is the minimum memory required.
    # Or, pass (request, limit) to additionally specify a hard limit in MiB.
    memory: Optional[Union[int, Tuple[int, int]]] = None,
    block_network: bool = False,  # Whether to block network access
    volumes: Dict[
        Union[str, os.PathLike], Union[_Volume, _CloudBucketMount]
    ] = {},  # Mount points for Modal Volumes and CloudBucketMounts
    pty_info: Optional[api_pb2.PTYInfo] = None,
    # List of ports to tunnel into the sandbox. Encrypted ports are tunneled with TLS.
    encrypted_ports: Sequence[int] = [],
    # List of ports to tunnel into the sandbox without encryption.
    unencrypted_ports: Sequence[int] = [],
    _experimental_scheduler_placement: Optional[
        SchedulerPlacement
    ] = None,  # Experimental controls over fine-grained scheduling (alpha).
    client: Optional[_Client] = None,
    _experimental_gpus: Sequence[GPU_T] = [],
) -> "_Sandbox":

from_id

@staticmethod
def from_id(sandbox_id: str, client: Optional[_Client] = None) -> "_Sandbox":

Construct a Sandbox from an id and look up the Sandbox result.

The ID of a Sandbox object can be accessed using .object_id.

wait

def wait(self, raise_on_termination: bool = True):

Wait for the Sandbox to finish running.

tunnels

def tunnels(self, timeout: int = 50) -> List[api_pb2.TunnelData]:

Get tunnel metadata for the sandbox.

Raises SandboxTimeoutError if the tunnels are not available after the timeout.

Returns a list of TunnelData objects, which contain the tunnel metadata.

terminate

def terminate(self):

Terminate Sandbox execution.

This is a no-op if the Sandbox has already finished running.

poll

def poll(self) -> Optional[int]:

Check if the Sandbox has finished running.

Returns None if the Sandbox is still running, else returns the exit code.

exec

def exec(self, *cmds: str, pty_info: Optional[api_pb2.PTYInfo] = None):

Execute a command in the Sandbox and return a ContainerProcess handle.

Usage

sandbox = modal.Sandbox.create("sleep", "infinity")

process = sandbox.exec("bash", "-c", "for i in $(seq 1 10); do echo foo $i; sleep 0.5; done")

for line in process.stdout:
    print(line)

stdout

@property
def stdout(self) -> _StreamReader:

StreamReader for the sandbox’s stdout stream.

stderr

@property
def stderr(self) -> _StreamReader:

StreamReader for the sandbox’s stderr stream.

stdin

@property
def stdin(self) -> _StreamWriter:

StreamWriter for the sandbox’s stdin stream.

returncode

@property
def returncode(self) -> Optional[int]:

Return code of the sandbox process if it has finished running, else None.