# app_run.py
"""Application run object for remote application execution."""
from contextlib import contextmanager
from typing import TYPE_CHECKING, Callable, Optional
if TYPE_CHECKING:
from .client_http import AppMeshClient
# Type alias for stdout callback: (data, position) -> None
OutputHandler = Optional[Callable[[str, int], None]]
[docs]
def print_output_handler(data: str, position: int) -> None:
"""Pre-built OutputHandler that prints data to stdout."""
print(data, end="", flush=True)
[docs]
class AppRun:
"""
Application run object for monitoring and retrieving results
of a remote application run initiated by `run_async()`.
"""
def __init__(self, client: "AppMeshClient", app_name: str, process_id: str):
self.app_name = app_name
"""Name of the application associated with this run."""
self.proc_uid = process_id
"""Unique process ID from `run_async()`."""
self._client = client
self._forward_to = client.forward_to
[docs]
@contextmanager
def forward_to(self):
"""
Context manager to temporarily override the client's `forward_to` setting.
Ensures operations during this run use the correct target server,
then restores the original setting.
"""
original_value = self._client.forward_to
self._client.forward_to = self._forward_to
try:
yield
finally:
self._client.forward_to = original_value
[docs]
def wait(self, stdout_handler: OutputHandler = None, timeout: int = 0) -> Optional[int]:
"""
Wait for the asynchronous run to complete with the saved forwarding target restored.
Args:
stdout_handler: optional callback ``(data, position) -> None`` invoked with each
chunk of stdout. Use ``print_output_handler`` for console output.
timeout: Maximum time to wait in seconds. 0 means wait indefinitely.
Returns:
Exit code if the process finishes successfully, or ``None`` on timeout/polling failure.
"""
with self.forward_to():
return self._client.wait_for_async_run(self, stdout_handler, timeout)