runtimepy.net.apps package

runtimepy.net.apps package#

Module contents#

A module aggregating commonly used network applications (that use the connection-arbiter technology).

class runtimepy.net.apps.AppInfo(logger: Logger | LoggerAdapter[Any], stack: AsyncExitStack, connections: MutableMapping[str, Connection], conn_manager: ConnectionManager, names: Namespace, stop: Event, config: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]], ports: dict[str, int], tui: TuiMixin, tasks: dict[str, PeriodicTask], task_manager: PeriodicTaskManager[Any], results: list[list[AppResult]], structs: dict[str, RuntimeStructBase], peers: dict[str, _RuntimepyPeer])[source]#

Bases: LoggerMixin

References provided to network applications.

async all_finalized() → None[source]#: Wait for all tasks and connections to be finalized.

config: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]#

config_param(key: str, default: Z, strict: bool = False) → Z[source]#: Attempt to get a configuration parameter.

conn_manager: ConnectionManager#

connections: MutableMapping[str, Connection]#

exceptions() → Iterator[Exception][source]#: Iterate over exceptions raised by the application.

logger: Logger | LoggerAdapter[Any]#

names: Namespace#

original_config() → dict[str, Any][source]#: Re-assemble a dictionary closer to the original configuration data (than the .config attribute).

peers: dict[str, _RuntimepyPeer]#

ports: dict[str, int]#

property raised_exception: bool#: Determine if the application raised any exception.

result(logger: Logger | LoggerAdapter[Any] = None) → bool[source]#: Get the overall boolean result for the application.

results: list[list[AppResult]]#

search(*names: str, pattern: str = '.*', kind: type[~runtimepy.net.arbiter.info.T] = <class 'runtimepy.net.connection.Connection'>) → Iterator[T][source]#: Get all connections that are matching a naming convention or are specific kind (or both).

search_structs(kind: type[W], pattern: str = '.*') → Iterator[W][source]#: Search for structs by type or name.

search_tasks(kind: type[V], pattern: str = '.*') → Iterator[V][source]#: Search for tasks by type or pattern.

single(*names: str, pattern: str = '.*', kind: type[~runtimepy.net.arbiter.info.T] = <class 'runtimepy.net.connection.Connection'>) → T[source]#: Search for a single node.

stack: AsyncExitStack#

stop: Event#

structs: dict[str, RuntimeStructBase]#

task_manager: PeriodicTaskManager[Any]#

tasks: dict[str, PeriodicTask]#

tui: TuiMixin#

with_new_logger(name: str) → AppInfo[source]#: Get a copy of this AppInfo instance, but with a new logger.

async runtimepy.net.apps.exception(app: AppInfo) → int[source]#: Waits for the stop signal to be set.

async runtimepy.net.apps.fail(app: AppInfo) → int[source]#: Waits for the stop signal to be set.

async runtimepy.net.apps.init_only(app: AppInfo) → int[source]#: A network application that doesn’t do anything.

async runtimepy.net.apps.noop(app: AppInfo) → int#: A network application that doesn’t do anything.

async runtimepy.net.apps.wait_for_stop(app: AppInfo) → int[source]#: Waits for the stop signal to be set.