Skip to main content

Module clients.sandbox.process.client

Classes​

ProcessClient​

class ProcessClient(connection_config: h2o_engine_manager.clients.connection_config.ConnectionConfig, verify_ssl: bool = True, ssl_ca_cert: Optional[str] = None)

ProcessClient manages process operations within a SandboxEngine.

Args​
connection_config
AIEM connection configuration object.
verify_ssl
Set to False to disable SSL certificate verification.
ssl_ca_cert
Path to a CA cert bundle with certificates of trusted CAs.

Methods​

create_process ​

def create_process(self, parent: str, process: h2o_engine_manager.clients.sandbox.process.ps.Process, process_id: str = '', auto_run: bool = False) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Create a new process in the sandbox engine.

Args​
parent : str
The parent SandboxEngine resource name. Format: "workspaces//sandboxEngines/"
process : Process
The Process to create.
process_id : str
Optional ID for the process. Must be 1-63 characters, lowercase alphanumeric or hyphen, start and end with alphanumeric. If not provided, a UUID is generated.
auto_run : bool
If True, starts execution immediately. If False, the process is created in STATE_PENDING and requires a separate start_process() call.
Returns​
Process
The created process.

get_process ​

def get_process(self, name: str) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Get a process by its resource name.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
Returns​
Process
The process.

list_processes ​

def list_processes(self, parent: str, page_size: int = 0, page_token: str = '', filter: str = '') ‑> tuple[typing.List[h2o_engine_manager.clients.sandbox.process.ps.Process], str]

List processes in a sandbox engine.

Args​
parent : str
The parent SandboxEngine resource name. Format: "workspaces//sandboxEngines/"
page_size : int
Maximum number of processes to return. If unspecified (or set to 0), the server default will be used.
page_token : str
Token for pagination. Leave unset to receive the initial page.
filter : str
Filter expression.
Returns​
tuple[List[Process], str]
A tuple of (processes, next_page_token).

read_output ​

def read_output(self, name: str, output_stream: str = 'OUTPUT_STREAM_COMBINED') ‑> bytes

Read the complete output from a process.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
output_stream : str
Which output stream to read. Options: - "OUTPUT_STREAM_STDOUT": Read only stdout - "OUTPUT_STREAM_STDERR": Read only stderr - "OUTPUT_STREAM_COMBINED": Read combined stdout and stderr (default)
Returns​
bytes
The output data as bytes.

send_signal ​

def send_signal(self, name: str, signal: int) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Send a signal to a running process.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
signal : int
The signal number to send (e.g., 9 for SIGKILL, 15 for SIGTERM).
Returns​
Process
The process after sending the signal.

start_process ​

def start_process(self, name: str) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Start a process that is in STATE_PENDING.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
Returns​
Process
The started process.

stream_output ​

def stream_output(self, name: str, skip_replay: bool = False) ‑> Iterator[bytes]

Stream output from a process as it becomes available.

This returns an iterator that yields output chunks as they are produced.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
skip_replay : bool
If True, skips replaying existing output and only streams new output. If False, replays all existing output first.
Yields​
bytes
Output chunks as they become available.

wait_process ​

def wait_process(self, name: str) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Wait for a process to complete (blocking call).

This method blocks until the process reaches a terminal state (STATE_SUCCEEDED or STATE_FAILED).

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
Returns​
Process
The completed process.

wait_process_by_polling ​

def wait_process_by_polling(self, name: str, timeout_seconds: Optional[int] = 300, poll_interval_seconds: int = 1) ‑> h2o_engine_manager.clients.sandbox.process.ps.Process

Wait for a process to reach a final state by polling (client-side blocking).

This method repeatedly calls get_process until the process reaches a final state (STATE_SUCCEEDED or STATE_FAILED) or the timeout is reached.

Args​
name : str
Process resource name. Format: "workspaces//sandboxEngines//processes/*"
timeout_seconds : Optional[int]
Maximum time to wait in seconds (default: 300). If None, waits indefinitely.
poll_interval_seconds : int
Time to wait between polling attempts in seconds (default: 1).
Returns​
Process
The process in its final state.
Raises​
TimeoutError
If the process does not reach a final state within the timeout.
Feedback