Skip to content

omnipy.shared.protocols.engine.base

Base protocols for Omnipy engine implementations.

CLASS DESCRIPTION
IsEngine

Protocol for execution engines configured with registry state.

IsEngine

Bases: Protocol


              flowchart BT
              omnipy.shared.protocols.engine.base.IsEngine[IsEngine]

              

              click omnipy.shared.protocols.engine.base.IsEngine href "" "omnipy.shared.protocols.engine.base.IsEngine"

Protocol for execution engines configured with registry state.

Returned by: Subclassed by: Used by:
METHOD DESCRIPTION
__init__
get_config_cls

Return the config class associated with this engine type.
set_config

Replace the active config and refresh config-dependent state.
set_registry

Attach or clear the run-state registry used for job state reporting.
ATTRIBUTE DESCRIPTION
config

Return the currently active engine configuration.

TYPE: IsJobRunnerConfig

registry

Return the registry currently used for run-state reporting.

TYPE: IsRunStateRegistry | None

Source code in src/omnipy/shared/protocols/engine/base.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@runtime_checkable
class IsEngine(Protocol):
    """Protocol for execution engines configured with registry state."""
    def __init__(self) -> None:
        ...

    @classmethod
    def get_config_cls(cls) -> Type[IsJobRunnerConfig]:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISENGINE_GET_CONFIG_CLS_SUMMARY}}
        #
        # {{ISENGINE_GET_CONFIG_CLS_DETAILS}}
        """Return the config class associated with this engine type.

        Args:
            cls: Engine subclass whose configuration class is being requested.

        Returns:
            Type[IsJobRunnerConfig]: Configuration class used to instantiate engine settings.
        """
        ...

    def set_config(self, config: IsJobRunnerConfig) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISENGINE_SET_CONFIG_SUMMARY}}
        #
        # {{ISENGINE_SET_CONFIG_DETAILS}}
        """Replace the active config and refresh config-dependent state.

        Args:
            config: Runtime configuration object for the engine.
        """
        ...

    def set_registry(self, registry: IsRunStateRegistry | None) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISENGINE_SET_REGISTRY_SUMMARY}}
        #
        # {{ISENGINE_SET_REGISTRY_DETAILS}}
        """Attach or clear the run-state registry used for job state reporting.

        Args:
            registry: Registry implementation, or ``None`` to disable state reporting.
        """
        ...

    @property
    def config(self) -> IsJobRunnerConfig:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISENGINE_CONFIG_SUMMARY}}
        #
        # {{ISENGINE_CONFIG_DETAILS}}
        """Return the currently active engine configuration.

        Returns:
            IsJobRunnerConfig: Active configuration object controlling engine behavior.
        """
        ...

    @property
    def registry(self) -> IsRunStateRegistry | None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISENGINE_REGISTRY_SUMMARY}}
        #
        # {{ISENGINE_REGISTRY_DETAILS}}
        """Return the registry currently used for run-state reporting.

        Returns:
            IsRunStateRegistry | None: Registry receiving job-state updates, or ``None`` when
                state reporting is disabled.
        """
        ...

config property 

config: IsJobRunnerConfig

Return the currently active engine configuration.

RETURNS DESCRIPTION
IsJobRunnerConfig

Active configuration object controlling engine behavior.

TYPE: IsJobRunnerConfig

registry property 

registry: IsRunStateRegistry | None

Return the registry currently used for run-state reporting.

RETURNS DESCRIPTION
IsRunStateRegistry | None

IsRunStateRegistry | None: Registry receiving job-state updates, or None when state reporting is disabled.

__init__ 

__init__() -> None
Source code in src/omnipy/shared/protocols/engine/base.py 
55
56
def __init__(self) -> None:
    ...

get_config_cls classmethod 

get_config_cls() -> Type[IsJobRunnerConfig]

Return the config class associated with this engine type.

PARAMETER DESCRIPTION
cls

Engine subclass whose configuration class is being requested.
RETURNS DESCRIPTION
Type[IsJobRunnerConfig]

Type[IsJobRunnerConfig]: Configuration class used to instantiate engine settings.
Source code in src/omnipy/shared/protocols/engine/base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def get_config_cls(cls) -> Type[IsJobRunnerConfig]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISENGINE_GET_CONFIG_CLS_SUMMARY}}
    #
    # {{ISENGINE_GET_CONFIG_CLS_DETAILS}}
    """Return the config class associated with this engine type.

    Args:
        cls: Engine subclass whose configuration class is being requested.

    Returns:
        Type[IsJobRunnerConfig]: Configuration class used to instantiate engine settings.
    """
    ...

set_config 

set_config(config: IsJobRunnerConfig) -> None

Replace the active config and refresh config-dependent state.

PARAMETER DESCRIPTION
config

Runtime configuration object for the engine.

TYPE: IsJobRunnerConfig

Source code in src/omnipy/shared/protocols/engine/base.py 
74
75
76
77
78
79
80
81
82
83
84
def set_config(self, config: IsJobRunnerConfig) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISENGINE_SET_CONFIG_SUMMARY}}
    #
    # {{ISENGINE_SET_CONFIG_DETAILS}}
    """Replace the active config and refresh config-dependent state.

    Args:
        config: Runtime configuration object for the engine.
    """
    ...

set_registry 

set_registry(registry: IsRunStateRegistry | None) -> None

Attach or clear the run-state registry used for job state reporting.

PARAMETER DESCRIPTION
registry

Registry implementation, or None to disable state reporting.

TYPE: IsRunStateRegistry | None

Source code in src/omnipy/shared/protocols/engine/base.py 
86
87
88
89
90
91
92
93
94
95
96
def set_registry(self, registry: IsRunStateRegistry | None) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISENGINE_SET_REGISTRY_SUMMARY}}
    #
    # {{ISENGINE_SET_REGISTRY_DETAILS}}
    """Attach or clear the run-state registry used for job state reporting.

    Args:
        registry: Registry implementation, or ``None`` to disable state reporting.
    """
    ...