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.