omnipy.engine.local

CLASS	DESCRIPTION
LocalRunner	Local job runner

LocalRunner

Bases: JobRunnerEngine

              flowchart BT
              omnipy.engine.local.LocalRunner[LocalRunner]
              omnipy.engine.job_runner.JobRunnerEngine[JobRunnerEngine]
              omnipy.engine._base.Engine[Engine]

                              omnipy.engine.job_runner.JobRunnerEngine --> omnipy.engine.local.LocalRunner
                                omnipy.engine._base.Engine --> omnipy.engine.job_runner.JobRunnerEngine
                



              click omnipy.engine.local.LocalRunner href "" "omnipy.engine.local.LocalRunner"
              click omnipy.engine.job_runner.JobRunnerEngine href "" "omnipy.engine.job_runner.JobRunnerEngine"
              click omnipy.engine._base.Engine href "" "omnipy.engine._base.Engine"
            

Local job runner

Used by:

• Reference Code reference omnipy hub  runtime  RuntimeObjects  local

METHOD	DESCRIPTION
__init__	Initialize engine config, registry holder, and subclass state.
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

Source code in src/omnipy/engine/local.py

class LocalRunner(JobRunnerEngine):
    """Local job runner"""
    supported_job_types = frozenset({
        JobType.TASK,
        JobType.LINEAR_FLOW,
        JobType.DAG_FLOW,
        JobType.FUNC_FLOW,
    })

    def _init_engine(self) -> None:
        """Initialize local-runner-specific engine state."""
        ...

    def _update_from_config(self) -> None:
        """Apply updates when local runner config changes."""
        ...

    @classmethod
    def get_config_cls(cls) -> Type[IsLocalRunnerConfig]:
        # %% 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.
        """
        return cast(Type[IsLocalRunnerConfig], LocalRunnerConfig)

    def _init_task(self, task: TaskRunSpec) -> object:
        ...

    def _run_task(self, state: Any, task: TaskRunSpec, *args, **kwargs) -> object:
        return task.create_default_run_callable()(*args, **kwargs)

    def _init_flow(self, flow: FlowRunSpec) -> object:
        ...

    def _run_flow(self, state: Any, flow: FlowRunSpec, *args, **kwargs) -> object:
        return flow.create_default_run_callable()(*args, **kwargs)

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 class-attribute instance-attribute

supported_job_types = frozenset(
    {JobType.TASK, JobType.LINEAR_FLOW, JobType.DAG_FLOW, JobType.FUNC_FLOW}
)

__init__

__init__() -> None

Initialize engine config, registry holder, and subclass state.

Source code in src/omnipy/engine/_base.py

def __init__(self) -> None:
    """Initialize engine config, registry holder, and subclass state."""
    config_cls = self.get_config_cls()
    self._config: IsJobRunnerConfig = config_cls()
    self._registry: IsRunStateRegistry | None = None

    self._init_engine()

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/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.
    """
    self._require_support(job_type)
    job_run_spec_cls, init_state, run_job = self._job_type_to_run_spec_and_funcs(job_type)

    self._apply_job_decorator(
        job,
        job_callback_accept_decorator,
        job_run_spec_cls,
        init_state,
        run_job,
    )

get_config_cls classmethod

get_config_cls() -> Type[IsLocalRunnerConfig]

Return the config class associated with this engine type.

PARAMETER	DESCRIPTION
cls	Engine subclass whose configuration class is being requested.

RETURNS	DESCRIPTION
Type[IsLocalRunnerConfig]	Type[IsJobRunnerConfig]: Configuration class used to instantiate engine settings.

Source code in src/omnipy/engine/local.py

@classmethod
def get_config_cls(cls) -> Type[IsLocalRunnerConfig]:
    # %% 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.
    """
    return cast(Type[IsLocalRunnerConfig], LocalRunnerConfig)

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/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.
    """
    self._config = config
    self._update_from_config()

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/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.
    """
    self._registry = registry

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/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.
    """
    return job_type in self.supported_job_types