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, command: str, args: Optional[List[str]] = None, working_directory: Optional[str] = None, environment_variables: Optional[Dict[str, str]] = None, secret_environment_variables: Optional[Dict[str, h2o_engine_manager.clients.sandbox.process.secret_environment_variable.SecretEnvironmentVariable]] = None, process_id: Optional[str] = None, 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/"
command : str
Required. The command to execute.
args : Optional[List[str]]
Optional arguments for the command.
working_directory : Optional[str]
Optional working directory for the process.
environment_variables : Optional[Dict[str, str]]
Optional environment variables.
secret_environment_variables (Optional[Dict[str, SecretEnvironmentVariable]]):
Optional environment variables from H2O Secure Store secrets.
process_id : Optional[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: Optional[int] = None, page_token: Optional[str] = None, filter: Optional[str] = None) ‑> 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 : Optional[int]
Maximum number of processes to return.
page_token : Optional[str]
Token for pagination.
filter : Optional[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.
Feedback