yera.events.runtime

App runtime execution and stream handling.

Symbols

class PyRuntimeExecutor — Execute an app in a separate process using an EventStream.
def stream_handler — Handle an event stream and render it to the console.

PyRuntimeExecutor

Execute an app in a separate process using an EventStream.

Methods

start — Start the app execution in a new process with an internally created EventStream.
stop — Stop the app process if it is running and clean up resources.
is_running — Check if the executor process is running.
__enter__ — Start execution with a fresh EventStream and return self.
__exit__ — Ensure the subprocess is terminated when leaving the context.

PyRuntimeExecutor.start

start() → None

Start the app execution in a new process with an internally created EventStream.

PyRuntimeExecutor.stop

stop() → None

Stop the app process if it is running and clean up resources.

PyRuntimeExecutor.is_running

is_running() → bool

Check if the executor process is running.

Returns

type: bool

True if process is alive and stream is available

PyRuntimeExecutor.__enter__

__enter__()

Start execution with a fresh EventStream and return self.

PyRuntimeExecutor.__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
)

Ensure the subprocess is terminated when leaving the context.

stream_handler

stream_handler(
    metadata: AppMetadata,
    console: Console | None = None,
    interactive: bool | None = None,
) → OutputEvent

Handle an event stream and render it to the console.

Parameters

metadata
type: AppMetadata

The metadata of the app to run.

console
type: Console | None = None

The console to render to.

interactive
type: bool | None = None

Whether the stdin is interactive.