omnipy.shared.protocols.hub.runtime

Protocols for runtime configuration, objects, and root logging.

CLASS	DESCRIPTION
IsRootLogObjects	Protocol for root logger helper objects.
IsRuntime	Protocol for the Omnipy runtime container.
IsRuntimeConfig	Protocol for the aggregated runtime configuration.
IsRuntimeObjects	Protocol for the instantiated objects owned by the runtime.

IsRootLogObjects

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.hub.runtime.IsRootLogObjects[IsRootLogObjects]

              

              click omnipy.shared.protocols.hub.runtime.IsRootLogObjects href "" "omnipy.shared.protocols.hub.runtime.IsRootLogObjects"
            

Protocol for root logger helper objects.

Returned by:

• Reference Code reference omnipy
  • hub  runtime  RuntimeObjects  root_log
  • shared protocols hub  runtime  IsRuntimeObjects  root_log

METHOD	DESCRIPTION
set_config	Replace root logging configuration and rebuild logging objects.

ATTRIBUTE	DESCRIPTION
config	Return the active root logging configuration. TYPE: IsRootLogConfig
file_handler	TYPE: RotatingFileHandler | None
formatter	TYPE: logging.Formatter | None
stderr_handler	TYPE: logging.StreamHandler | None
stdout_handler	TYPE: logging.StreamHandler | None

Source code in src/omnipy/shared/protocols/hub/runtime.py

@runtime_checkable
class IsRootLogObjects(Protocol):
    """Protocol for root logger helper objects."""

    formatter: logging.Formatter | None = None
    stdout_handler: logging.StreamHandler | None = None
    stderr_handler: logging.StreamHandler | None = None
    file_handler: RotatingFileHandler | None = None

    def set_config(self, config: IsRootLogConfig) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISROOTLOGOBJECTS_SET_CONFIG_SUMMARY}}
        #
        # {{ISROOTLOGOBJECTS_SET_CONFIG_DETAILS}}
        """Replace root logging configuration and rebuild logging objects.

        Args:
            config: New root logger configuration to apply.
        """
        ...

    @property
    def config(self) -> IsRootLogConfig:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISROOTLOGOBJECTS_CONFIG_SUMMARY}}
        #
        # {{ISROOTLOGOBJECTS_CONFIG_DETAILS}}
        """Return the active root logging configuration.

        Returns:
            IsRootLogConfig: Configuration currently used to construct formatter and handlers.
        """
        ...

config property

config: IsRootLogConfig

Return the active root logging configuration.

RETURNS	DESCRIPTION
IsRootLogConfig	Configuration currently used to construct formatter and handlers. TYPE: IsRootLogConfig

file_handler class-attribute instance-attribute

file_handler: RotatingFileHandler | None = None

formatter class-attribute instance-attribute

formatter: logging.Formatter | None = None

stderr_handler class-attribute instance-attribute

stderr_handler: logging.StreamHandler | None = None

stdout_handler class-attribute instance-attribute

stdout_handler: logging.StreamHandler | None = None

set_config

set_config(config: IsRootLogConfig) -> None

Replace root logging configuration and rebuild logging objects.

PARAMETER	DESCRIPTION
config	New root logger configuration to apply. TYPE: IsRootLogConfig

Source code in src/omnipy/shared/protocols/hub/runtime.py

def set_config(self, config: IsRootLogConfig) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISROOTLOGOBJECTS_SET_CONFIG_SUMMARY}}
    #
    # {{ISROOTLOGOBJECTS_SET_CONFIG_DETAILS}}
    """Replace root logging configuration and rebuild logging objects.

    Args:
        config: New root logger configuration to apply.
    """
    ...

IsRuntime

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.hub.runtime.IsRuntime[IsRuntime]

              

              click omnipy.shared.protocols.hub.runtime.IsRuntime href "" "omnipy.shared.protocols.hub.runtime.IsRuntime"
            

Protocol for the Omnipy runtime container.

Returned by:

• Reference Code reference
  •  omnipy  runtime
  • omnipy hub  runtime  runtime

METHOD	DESCRIPTION
reset_subscriptions	Reset runtime subscriptions between config and runtime objects.

ATTRIBUTE	DESCRIPTION
config	TYPE: IsRuntimeConfig
objects	TYPE: IsRuntimeObjects

Source code in src/omnipy/shared/protocols/hub/runtime.py

@runtime_checkable
class IsRuntime(Protocol):
    """Protocol for the Omnipy runtime container."""

    config: IsRuntimeConfig
    objects: IsRuntimeObjects

    def reset_subscriptions(self):
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIME_RESET_SUBSCRIPTIONS_SUMMARY}}
        #
        # {{ISRUNTIME_RESET_SUBSCRIPTIONS_DETAILS}}
        """Reset runtime subscriptions between config and runtime objects.

        This method rebuilds the callback graph that keeps configuration, engines, registries,
        logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects
        manually.

        Raises:
            AssertionError: If a Jupyter UI is detected but reactive objects are unexpectedly
                missing.
        """
        ...

config instance-attribute

config: IsRuntimeConfig

objects instance-attribute

objects: IsRuntimeObjects

reset_subscriptions

reset_subscriptions()

Reset runtime subscriptions between config and runtime objects.

This method rebuilds the callback graph that keeps configuration, engines, registries, logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects manually.

RAISES	DESCRIPTION
AssertionError	If a Jupyter UI is detected but reactive objects are unexpectedly missing.

Source code in src/omnipy/shared/protocols/hub/runtime.py

def reset_subscriptions(self):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIME_RESET_SUBSCRIPTIONS_SUMMARY}}
    #
    # {{ISRUNTIME_RESET_SUBSCRIPTIONS_DETAILS}}
    """Reset runtime subscriptions between config and runtime objects.

    This method rebuilds the callback graph that keeps configuration, engines, registries,
    logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects
    manually.

    Raises:
        AssertionError: If a Jupyter UI is detected but reactive objects are unexpectedly
            missing.
    """
    ...

IsRuntimeConfig

Bases: IsConfigBase, Protocol

              flowchart BT
              omnipy.shared.protocols.hub.runtime.IsRuntimeConfig[IsRuntimeConfig]
              omnipy.shared.protocols.config.IsConfigBase[IsConfigBase]
              omnipy.shared.protocols.util.IsDataPublisher[IsDataPublisher]

                              omnipy.shared.protocols.config.IsConfigBase --> omnipy.shared.protocols.hub.runtime.IsRuntimeConfig
                                omnipy.shared.protocols.util.IsDataPublisher --> omnipy.shared.protocols.config.IsConfigBase
                



              click omnipy.shared.protocols.hub.runtime.IsRuntimeConfig href "" "omnipy.shared.protocols.hub.runtime.IsRuntimeConfig"
              click omnipy.shared.protocols.config.IsConfigBase href "" "omnipy.shared.protocols.config.IsConfigBase"
              click omnipy.shared.protocols.util.IsDataPublisher href "" "omnipy.shared.protocols.util.IsDataPublisher"
            

Protocol for the aggregated runtime configuration.

Returned by:

• Reference Code reference omnipy
  • hub  runtime  Runtime  config
  • shared protocols hub  runtime  IsRuntime  config

METHOD	DESCRIPTION
as_model
deepcopy	Return a deep-copied publisher instance.
default_repr_to_terminal_str	Render the config for a terminal-style user interface.
reset_to_defaults	Reset all runtime configuration sections to their default values.
subscribe	Subscribe to general publisher updates.
subscribe_attr	Subscribe to updates for one named attribute.
unsubscribe_all	Remove every registered subscriber.

ATTRIBUTE	DESCRIPTION
data	TYPE: IsDataConfig
engine	TYPE: IsEngineConfig
job	TYPE: IsJobConfig
root_log	TYPE: IsRootLogConfig

Source code in src/omnipy/shared/protocols/hub/runtime.py

@runtime_checkable
class IsRuntimeConfig(IsConfigBase, Protocol):
    """Protocol for the aggregated runtime configuration."""

    data: IsDataConfig
    engine: IsEngineConfig
    job: IsJobConfig
    root_log: IsRootLogConfig

    def reset_to_defaults(self) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_SUMMARY}}
        #
        # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_DETAILS}}
        """Reset all runtime configuration sections to their default values.

        Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime
        subscriptions when the config is attached to a runtime object.
        """
        ...

data instance-attribute

data: IsDataConfig

engine instance-attribute

engine: IsEngineConfig

job instance-attribute

job: IsJobConfig

root_log instance-attribute

root_log: IsRootLogConfig

as_model

as_model() -> IsModel[Any]

Source code in src/omnipy/shared/protocols/config.py

def as_model(self) -> 'IsModel[Any]':
    ...

deepcopy

deepcopy() -> Self

Return a deep-copied publisher instance.

RETURNS	DESCRIPTION
Self	Deep copy of the current publisher object. TYPE: Self

Source code in src/omnipy/shared/protocols/util.py

def deepcopy(self) -> Self:
    """Return a deep-copied publisher instance.

    Returns:
        Self: Deep copy of the current publisher object.
    """
    ...

default_repr_to_terminal_str

default_repr_to_terminal_str(ui_type: TerminalOutputUserInterfaceType.Literals) -> str

Render the config for a terminal-style user interface.

PARAMETER	DESCRIPTION
ui_type	Terminal frontend variant requesting the representation. TYPE: TerminalOutputUserInterfaceType.Literals

RETURNS	DESCRIPTION
str	Terminal-friendly string representation of the config. TYPE: str

Source code in src/omnipy/shared/protocols/config.py

def default_repr_to_terminal_str(
    self,
    ui_type: TerminalOutputUserInterfaceType.Literals,
) -> str:
    """Render the config for a terminal-style user interface.

    Args:
        ui_type: Terminal frontend variant requesting the representation.

    Returns:
        str: Terminal-friendly string representation of the config.
    """
    ...

reset_to_defaults

reset_to_defaults() -> None

Reset all runtime configuration sections to their default values.

Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime subscriptions when the config is attached to a runtime object.

Source code in src/omnipy/shared/protocols/hub/runtime.py

def reset_to_defaults(self) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_SUMMARY}}
    #
    # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_DETAILS}}
    """Reset all runtime configuration sections to their default values.

    Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime
    subscriptions when the config is attached to a runtime object.
    """
    ...

subscribe

subscribe(callback_fun: Callable[..., None], do_callback: bool = True) -> None

Subscribe to general publisher updates.

PARAMETER	DESCRIPTION
callback_fun	Callback invoked when the publisher changes. TYPE: Callable[..., None]
do_callback	Whether to invoke the callback immediately after subscribing. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/util.py

def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
    """Subscribe to general publisher updates.

    Args:
        callback_fun: Callback invoked when the publisher changes.
        do_callback: Whether to invoke the callback immediately after subscribing.
    """
    ...

subscribe_attr

subscribe_attr(attr_name: str, callback_fun: Callable[..., None])

Subscribe to updates for one named attribute.

PARAMETER	DESCRIPTION
attr_name	Name of the published attribute to observe. TYPE: str
callback_fun	Callback invoked when that attribute changes. TYPE: Callable[..., None]

Source code in src/omnipy/shared/protocols/util.py

def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
    """Subscribe to updates for one named attribute.

    Args:
        attr_name: Name of the published attribute to observe.
        callback_fun: Callback invoked when that attribute changes.
    """
    ...

unsubscribe_all

unsubscribe_all() -> None

Remove every registered subscriber.

Source code in src/omnipy/shared/protocols/util.py

def unsubscribe_all(self) -> None:
    """Remove every registered subscriber."""
    ...

IsRuntimeObjects

Bases: IsDataPublisher, Protocol

              flowchart BT
              omnipy.shared.protocols.hub.runtime.IsRuntimeObjects[IsRuntimeObjects]
              omnipy.shared.protocols.util.IsDataPublisher[IsDataPublisher]

                              omnipy.shared.protocols.util.IsDataPublisher --> omnipy.shared.protocols.hub.runtime.IsRuntimeObjects
                


              click omnipy.shared.protocols.hub.runtime.IsRuntimeObjects href "" "omnipy.shared.protocols.hub.runtime.IsRuntimeObjects"
              click omnipy.shared.protocols.util.IsDataPublisher href "" "omnipy.shared.protocols.util.IsDataPublisher"
            

Protocol for the instantiated objects owned by the runtime.

Returned by:

• Reference Code reference omnipy
  • hub  runtime  Runtime  objects
  • shared protocols hub  runtime  IsRuntime  objects

METHOD	DESCRIPTION
deepcopy	Return a deep-copied publisher instance.
setup_reactive	Create or remove reactive UI helpers for the detected interface.
subscribe	Subscribe to general publisher updates.
subscribe_attr	Subscribe to updates for one named attribute.
unsubscribe_all	Remove every registered subscriber.

ATTRIBUTE	DESCRIPTION
data_class_creator	TYPE: IsDataClassCreator
job_creator	TYPE: IsJobConfigHolder
local	TYPE: IsEngine
prefect	TYPE: IsEngine
reactive	TYPE: IsReactiveObjects | None
registry	TYPE: IsRunStateRegistry
root_log	TYPE: IsRootLogObjects
serializers	TYPE: IsSerializerRegistry

Source code in src/omnipy/shared/protocols/hub/runtime.py

@runtime_checkable
class IsRuntimeObjects(IsDataPublisher, Protocol):
    """Protocol for the instantiated objects owned by the runtime."""

    job_creator: IsJobConfigHolder
    data_class_creator: IsDataClassCreator
    reactive: IsReactiveObjects | None
    local: IsEngine
    prefect: IsEngine
    registry: IsRunStateRegistry
    serializers: IsSerializerRegistry
    root_log: IsRootLogObjects

    def setup_reactive(self, ui_type: UserInterfaceType.Literals) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_SUMMARY}}
        #
        # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_DETAILS}}
        """Create or remove reactive UI helpers for the detected interface.

        Args:
            ui_type: Detected user-interface type for the current runtime.
        """
        ...

data_class_creator instance-attribute

data_class_creator: IsDataClassCreator

job_creator instance-attribute

job_creator: IsJobConfigHolder

local instance-attribute

local: IsEngine

prefect instance-attribute

prefect: IsEngine

reactive instance-attribute

reactive: IsReactiveObjects | None

registry instance-attribute

registry: IsRunStateRegistry

root_log instance-attribute

root_log: IsRootLogObjects

serializers instance-attribute

serializers: IsSerializerRegistry

deepcopy

deepcopy() -> Self

Return a deep-copied publisher instance.

RETURNS	DESCRIPTION
Self	Deep copy of the current publisher object. TYPE: Self

Source code in src/omnipy/shared/protocols/util.py

def deepcopy(self) -> Self:
    """Return a deep-copied publisher instance.

    Returns:
        Self: Deep copy of the current publisher object.
    """
    ...

setup_reactive

setup_reactive(ui_type: UserInterfaceType.Literals) -> None

Create or remove reactive UI helpers for the detected interface.

PARAMETER	DESCRIPTION
ui_type	Detected user-interface type for the current runtime. TYPE: UserInterfaceType.Literals

Source code in src/omnipy/shared/protocols/hub/runtime.py

def setup_reactive(self, ui_type: UserInterfaceType.Literals) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_SUMMARY}}
    #
    # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_DETAILS}}
    """Create or remove reactive UI helpers for the detected interface.

    Args:
        ui_type: Detected user-interface type for the current runtime.
    """
    ...

subscribe

subscribe(callback_fun: Callable[..., None], do_callback: bool = True) -> None

Subscribe to general publisher updates.

PARAMETER	DESCRIPTION
callback_fun	Callback invoked when the publisher changes. TYPE: Callable[..., None]
do_callback	Whether to invoke the callback immediately after subscribing. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/util.py

def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
    """Subscribe to general publisher updates.

    Args:
        callback_fun: Callback invoked when the publisher changes.
        do_callback: Whether to invoke the callback immediately after subscribing.
    """
    ...

subscribe_attr

subscribe_attr(attr_name: str, callback_fun: Callable[..., None])

Subscribe to updates for one named attribute.

PARAMETER	DESCRIPTION
attr_name	Name of the published attribute to observe. TYPE: str
callback_fun	Callback invoked when that attribute changes. TYPE: Callable[..., None]

Source code in src/omnipy/shared/protocols/util.py

def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
    """Subscribe to updates for one named attribute.

    Args:
        attr_name: Name of the published attribute to observe.
        callback_fun: Callback invoked when that attribute changes.
    """
    ...

unsubscribe_all

unsubscribe_all() -> None

Remove every registered subscriber.

Source code in src/omnipy/shared/protocols/util.py

def unsubscribe_all(self) -> None:
    """Remove every registered subscriber."""
    ...