runtimepy.net.arbiter.imports package

Submodules

runtimepy.net.arbiter.imports.util module

Utility interfaces for arbiter runtime-import mechanisms.

runtimepy.net.arbiter.imports.util.get_apps(module_path: str | list[str] | None, wait_for_stop: bool = False) → list[list[Callable[[AppInfo], Awaitable[int]]]]

Attempt to update the application method from the provided string.

Module contents

A module extending capability of the connection arbiter using Python import machinery.

class runtimepy.net.arbiter.imports.ImportConnectionArbiter(manager: ConnectionManager = None, stop_sig: Event = None, namespace: Namespace = None, logger: Logger | LoggerAdapter[Any] = None, app: Callable[[AppInfo], Awaitable[int]] | list[Callable[[AppInfo], Awaitable[int]]] = None, config: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]] = None, window: Any | None = None, metrics_poller_task: bool = True)

Bases: FactoryConnectionArbiter, TaskConnectionArbiter

A class implementing extensions to the connection arbiter for working with arbitrary Python modules.

factory_process(factory: str, name: str, top_level: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]) → bool

Register a runtime process.

factory_struct(factory: str, name: str, config: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]) → bool

Register a runtime structure from factory and name.

register_module_factory(module_path: str, *namespaces: str, **kwargs) → bool

Attempt to register a factory class based on its module path.

register_peer_factory(factory: type[RuntimepyPeer], *namespaces: str) → bool

Attempt to register a subprocess peer factory.

register_struct_factory(factory: type[RuntimeStruct], *namespaces: str) → bool

Attempt to register a struct factory.