Processes

Run a command to completion with run or sh, or spawn a SpacesdProcess to stream its output and write its input.

Python programs usually use the high-level API instead: sb.shell, sb.terminal.

Run processes#

Methods of SpacesdClient.

SpacesdClient.run#

Runs a command to completion.

async def run(self, command: SpacesdCommand) -> ProcessOutput

Returns ProcessOutput · Async · Raises CuaError

Example

import asyncio
 
import cua
 
async def main():
    guest = await cua.embedded().spacesd(URL, TOKEN)  # or: await sandbox.spacesd(None)
    out = await guest.run(cua.SpacesdCommand(program="echo", args=["hello"]))
    print([out.exit.success, out.stdout])
    # [True, b'hello\n']
 
asyncio.run(main())

SpacesdClient.sh#

Runs line with /bin/sh -c to completion.

async def sh(self, line: str, timeout_ms: Optional[int]) -> ProcessOutput

Returns ProcessOutput · Async · Raises CuaError

SpacesdClient.spawn#

Starts a command and returns its handle.

async def spawn(self, command: SpacesdCommand) -> SpacesdProcess

Returns SpacesdProcess · Async · Raises CuaError

Example

import asyncio
 
import cua
 
async def main():
    guest = await cua.embedded().spacesd(URL, TOKEN)
    proc = await guest.spawn(cua.SpacesdCommand(program="cat", args=[], stdin=True))
    await proc.write_stdin(b"xyz")
    await proc.close_stdin()
    print([(await proc.wait()).stdout])
    # [b'xyz']
 
asyncio.run(main())

SpacesdClient.attach#

Attaches to a running process by pid or tag.

async def attach(self, pid: Optional[int], tag: Optional[str], replay: ReplayMode) -> SpacesdProcess

Returns SpacesdProcess · Async · Raises CuaError

SpacesdClient.list_processes#

Guest processes as proto3 JSON ([ProcessInfo]).

async def list_processes(self, include_exited: bool) -> str

Returns String · Async · Raises CuaError

SpacesdProcess#

A guest process, from spawn or attach.

Returned by SpacesdClient.attach, SpacesdClient.spawn.

SpacesdProcess.close_stdin#

Closes stdin.

async def close_stdin(self) -> None

Async · Raises CuaError

SpacesdProcess.detach#

Stops receiving output without affecting the process.

async def detach(self) -> None

Async

SpacesdProcess.kill#

SIGKILL.

async def kill(self) -> None

Async · Raises CuaError

SpacesdProcess.next_event#

Next output chunk or the exit; None after the exit was returned. Reconnects transparently from the last seen offset.

async def next_event(self) -> Optional[ProcessEvent]

Returns Option<ProcessEvent> · Async · Raises CuaError

SpacesdProcess.resize#

Resizes the PTY.

async def resize(self, cols: int, rows: int) -> None

Async · Raises CuaError

SpacesdProcess.signal#

Sends a signal by name (term, kill, SIGINT, ...).

async def signal(self, signal: str) -> None

Async · Raises CuaError

SpacesdProcess.wait#

Collects the remaining output until exit.

async def wait(self) -> ProcessOutput

Returns ProcessOutput · Async · Raises CuaError

SpacesdProcess.write_pty#

Writes PTY keystrokes.

async def write_pty(self, data: bytes) -> None

Async · Raises CuaError

SpacesdProcess.write_stdin#

Writes stdin (PTY input for PTY processes), exactly once across retries.

async def write_stdin(self, data: bytes) -> None

Async · Raises CuaError

SpacesdCommand record#

A command to run in the guest.

ProcessOutput record#

Collected output of a finished process.

Returned by Sandbox.guest_sh, SpacesdClient.run, SpacesdClient.sh, SpacesdProcess.wait.

ExitInfo record#

How a process ended.

ProcessEvent record#

One output chunk or the exit.

Returned by SpacesdProcess.next_event.

ProcessEventKind enum#

Kind of a process event.

ProcessEventKind.STDOUT
ProcessEventKind.STDERR
ProcessEventKind.PTY
ProcessEventKind.EXIT

PtySize record#

A PTY size.

ReplayMode enum#

Scrollback replay when attaching.

ReplayMode.NONE()
ReplayMode.ALL()
ReplayMode.LAST_BYTES(n: int)
ReplayMode.FROM_OFFSET(offset: int)