omnipy.shared.protocols.engine.job_runner

Protocols for engines that decorate runnable jobs.

CLASS	DESCRIPTION
IsJobRunnerEngine

IsJobRunnerEngine

Bases: IsEngine, Protocol

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

                              omnipy.shared.protocols.engine.base.IsEngine --> omnipy.shared.protocols.engine.job_runner.IsJobRunnerEngine
                


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

METHOD	DESCRIPTION
__init__
apply_job_decorator	Attach the engine's execution decorator to a job callback endpoint.
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.
supports	Return whether the engine can initialize and run job_type jobs.

ATTRIBUTE	DESCRIPTION
config	Return the currently active engine configuration. TYPE: IsJobRunnerConfig
registry	Return the registry currently used for run-state reporting. TYPE: IsRunStateRegistry | None
supported_job_types	TYPE: frozenset[JobType.Literals]

Source code in src/omnipy/shared/protocols/engine/job_runner.py

@runtime_checkable
class IsJobRunnerEngine(IsEngine, Protocol):
    """"""
    supported_job_types: frozenset[JobType.Literals]

    def supports(self, job_type: JobType.Literals) -> bool:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISJOBRUNNERENGINE_SUPPORTS_SUMMARY}}
        #
        # {{ISJOBRUNNERENGINE_SUPPORTS_DETAILS}}
        """Return whether the engine can initialize and run ``job_type`` jobs.

        Args:
            job_type: Job category to test.

        Returns:
            bool: ``True`` when ``job_type`` is supported by the engine.
        """
        ...

    def apply_job_decorator(
        self,
        job_type: JobType.Literals,
        job: IsFuncArgJob,
        job_callback_accept_decorator: Callable,
    ) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISJOBRUNNERENGINE_APPLY_JOB_DECORATOR_SUMMARY}}
        #
        # {{ISJOBRUNNERENGINE_APPLY_JOB_DECORATOR_DETAILS}}
        """Attach the engine's execution decorator to a job callback endpoint.

        Args:
            job_type: Job category selecting the run behavior.
            job: Job instance being prepared for execution.
            job_callback_accept_decorator: Consumer that accepts the engine-provided
                decorator.
        """
        ...

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.

supported_job_types instance-attribute

supported_job_types: frozenset[JobType.Literals]

__init__

__init__() -> None

Source code in src/omnipy/shared/protocols/engine/base.py

def __init__(self) -> None:
    ...

apply_job_decorator

apply_job_decorator(
    job_type: JobType.Literals, job: IsFuncArgJob, job_callback_accept_decorator: Callable
) -> None

Attach the engine's execution decorator to a job callback endpoint.

PARAMETER	DESCRIPTION
job_type	Job category selecting the run behavior. TYPE: JobType.Literals
job	Job instance being prepared for execution. TYPE: IsFuncArgJob
job_callback_accept_decorator	Consumer that accepts the engine-provided decorator. TYPE: Callable

Source code in src/omnipy/shared/protocols/engine/job_runner.py

def apply_job_decorator(
    self,
    job_type: JobType.Literals,
    job: IsFuncArgJob,
    job_callback_accept_decorator: Callable,
) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBRUNNERENGINE_APPLY_JOB_DECORATOR_SUMMARY}}
    #
    # {{ISJOBRUNNERENGINE_APPLY_JOB_DECORATOR_DETAILS}}
    """Attach the engine's execution decorator to a job callback endpoint.

    Args:
        job_type: Job category selecting the run behavior.
        job: Job instance being prepared for execution.
        job_callback_accept_decorator: Consumer that accepts the engine-provided
            decorator.
    """
    ...

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

@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

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

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.
    """
    ...

supports

supports(job_type: JobType.Literals) -> bool

Return whether the engine can initialize and run job_type jobs.

PARAMETER	DESCRIPTION
job_type	Job category to test. TYPE: JobType.Literals

RETURNS	DESCRIPTION
bool	True when job_type is supported by the engine. TYPE: bool

Source code in src/omnipy/shared/protocols/engine/job_runner.py

def supports(self, job_type: JobType.Literals) -> bool:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBRUNNERENGINE_SUPPORTS_SUMMARY}}
    #
    # {{ISJOBRUNNERENGINE_SUPPORTS_DETAILS}}
    """Return whether the engine can initialize and run ``job_type`` jobs.

    Args:
        job_type: Job category to test.

    Returns:
        bool: ``True`` when ``job_type`` is supported by the engine.
    """
    ...