modal.sandbox

modal.sandbox.LogsReader

class LogsReader(object)

Provides an interface to buffer and fetch logs from a sandbox stream (stdout or stderr).

As an asynchronous iterable, the object supports the async for statement.

Usage

from modal import Sandbox

sandbox = Sandbox.create(
    "bash",
    "-c",
    "for i in $(seq 1 10); do echo foo; sleep 0.1; done"
)
for message in sandbox.stdout:
    print(f"Message: {message}")

read

def read(self) -> str:

Fetch and return contents of the entire stream. If EOF was received, return an empty string.

Usage

from modal import Sandbox

sandbox = Sandbox.create("echo", "hello")
sandbox.wait()

print(sandbox.stdout.read())

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,
    _allow_background_volume_commits: None = None,
    _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.

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.

stdout

@property
def stdout(self) -> _LogsReader:

LogsReader for the sandbox’s stdout stream.

stderr

@property
def stderr(self) -> _LogsReader:

LogsReader 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.

modal.sandbox.StreamWriter

class StreamWriter(object)

Provides an interface to buffer and write logs to a sandbox stream (stdin).

def __init__(self, sandbox_id: str, client: _Client):

write

def write(self, data: Union[bytes, bytearray, memoryview]):

Writes data to stream’s internal buffer, but does not drain/flush the write.

This method needs to be used along with the drain() method which flushes the buffer.

Usage

from modal import Sandbox

sandbox = Sandbox.create(
    "bash",
    "-c",
    "while read line; do echo $line; done",
)
sandbox.stdin.write(b"foo\n")
sandbox.stdin.write(b"bar\n")
sandbox.stdin.write_eof()

sandbox.stdin.drain()
sandbox.wait()

write_eof

def write_eof(self):

Closes the write end of the stream after the buffered write data is drained. If the sandbox process was blocked on input, it will become unblocked after write_eof().

This method needs to be used along with the drain() method which flushes the EOF to the process.

drain

def drain(self):

Flushes the write buffer and EOF to the running Sandbox process.