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
.