Skip to content

omnipy.engine.job_runner

CLASS DESCRIPTION
JobRunDef

Describe the run-spec class and hook names used for one job type.
JobRunnerEngine

Base class for job runner engine implementations
ATTRIBUTE DESCRIPTION
ExecRunHookName

FLOW_RUN_HOOKS

TYPE: JobRunHookNames

InitRunHookName

JOB_TYPE_TO_RUN_DEF

TYPE: dict[JobType.Literals, JobRunDef]

JobRunHookNames

TASK_RUN_HOOKS

TYPE: JobRunHookNames

ExecRunHookName module-attribute 

ExecRunHookName = Literal['_run_task', '_run_flow']
Returned by: Used by:

FLOW_RUN_HOOKS module-attribute 

FLOW_RUN_HOOKS: JobRunHookNames = ('_init_flow', '_run_flow')
Used by:

InitRunHookName module-attribute 

InitRunHookName = Literal['_init_task', '_init_flow']
Returned by: Used by:

JOB_TYPE_TO_RUN_DEF module-attribute 

JOB_TYPE_TO_RUN_DEF: dict[JobType.Literals, JobRunDef] = {
    JobType.TASK: JobRunDef(TaskRunSpec, TASK_RUN_HOOKS[0], TASK_RUN_HOOKS[1]),
    JobType.LINEAR_FLOW: JobRunDef(LinearFlowRunSpec, FLOW_RUN_HOOKS[0], FLOW_RUN_HOOKS[1]),
    JobType.DAG_FLOW: JobRunDef(DagFlowRunSpec, FLOW_RUN_HOOKS[0], FLOW_RUN_HOOKS[1]),
    JobType.FUNC_FLOW: JobRunDef(FuncFlowRunSpec, FLOW_RUN_HOOKS[0], FLOW_RUN_HOOKS[1]),
}

JobRunHookNames module-attribute 

JobRunHookNames = tuple[InitRunHookName, ExecRunHookName]
Returned by:

TASK_RUN_HOOKS module-attribute 

TASK_RUN_HOOKS: JobRunHookNames = ('_init_task', '_run_task')
Used by:

JobRunDef

Bases: NamedTuple


              flowchart BT
              omnipy.engine.job_runner.JobRunDef[JobRunDef]

              

              click omnipy.engine.job_runner.JobRunDef href "" "omnipy.engine.job_runner.JobRunDef"

Describe the run-spec class and hook names used for one job type.

Returned by: Used by:
ATTRIBUTE DESCRIPTION
init_hook_name

TYPE: InitRunHookName

run_hook_name

TYPE: ExecRunHookName

run_spec

TYPE: type[JobRunSpec]

Source code in src/omnipy/engine/job_runner.py 
20
21
22
23
24
25
class JobRunDef(NamedTuple):
    """Describe the run-spec class and hook names used for one job type."""

    run_spec: type[JobRunSpec]
    init_hook_name: InitRunHookName
    run_hook_name: ExecRunHookName

init_hook_name instance-attribute 

init_hook_name: InitRunHookName

run_hook_name instance-attribute 

run_hook_name: ExecRunHookName

run_spec instance-attribute 

run_spec: type[JobRunSpec]

JobRunnerEngine

Bases: Engine, ABC


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

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


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

Base class for job runner engine implementations

Subclassed by:
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

TYPE: frozenset[JobType.Literals]

Source code in src/omnipy/engine/job_runner.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class JobRunnerEngine(Engine, ABC):
    """Base class for job runner engine implementations"""
    supported_job_types: ClassVar[frozenset[JobType.Literals]] = frozenset()

    @classmethod
    def __init_subclass__(cls) -> None:
        super().__init_subclass__()

        assert cls.supported_job_types, ('`supported_job_types` must be set '
                                         'on JobRunnerEngine subclasses')

        def _require_hook_override(hook_name: str, support_desc: str) -> None:
            if getattr(cls, hook_name) is getattr(JobRunnerEngine, hook_name):
                raise TypeError(f'JobRunnerEngine subclass {cls.__name__} supports '
                                f'{support_desc} but does not override {hook_name}')

        unknown_job_types = set(cls.supported_job_types) - set(JOB_TYPE_TO_RUN_DEF)
        if unknown_job_types:
            raise TypeError(f'JobRunnerEngine subclass {cls.__name__} has unknown supported job '
                            f'types: {sorted(unknown_job_types)}')

        required_run_hooks = {
            hook_name for job_type in cls.supported_job_types for hook_name in (
                JOB_TYPE_TO_RUN_DEF[job_type].init_hook_name,
                JOB_TYPE_TO_RUN_DEF[job_type].run_hook_name,)
        }
        for hook_name in required_run_hooks:
            _require_hook_override(hook_name, 'supported jobs')

    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

    def _require_support(self, job_type: JobType.Literals) -> None:
        if not self.supports(job_type):
            raise RuntimeError(f'JobRunnerEngine "{self.__class__.__name__}" does not '
                               f'support job type: {job_type}')

    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,
        )

    def _job_type_to_run_spec_and_funcs(
            self, job_type: JobType.Literals) -> tuple[type[JobRunSpec], Callable, Callable]:
        if job_type not in JOB_TYPE_TO_RUN_DEF:
            raise ValueError(f'Unknown job type: {job_type}')

        job_run_def = JOB_TYPE_TO_RUN_DEF[job_type]
        job_run_spec, init_hook_name, run_hook_name = job_run_def
        return job_run_spec, getattr(self, init_hook_name), getattr(self, run_hook_name)

    def _register_job_state(self, job: IsFuncArgJob, state: RunState.Literals) -> None:
        if self._registry:
            self._registry.set_job_state(job, state)

    def _apply_job_decorator(
        self,
        job: IsFuncArgJob,
        job_callback_accept_decorator: Callable,
        job_run_spec_cls: type[JobRunSpec],
        init_state: Callable,
        run_job: Callable,
    ) -> None:
        def _job_decorator(call_func: Callable) -> Callable:
            job_run_spec = job_run_spec_cls(job, call_func)
            self._register_job_state(job, RunState.INITIALIZED)
            state = init_state(job_run_spec)

            def _job_runner_call_func(*args: object, **kwargs: object) -> Any:
                self._register_job_state(job, RunState.RUNNING)
                job_result = run_job(state, job_run_spec, *args, **kwargs)
                return self._decorate_result_with_job_finalization_detector(job, job_result)

            return _job_runner_call_func

        job_callback_accept_decorator(_job_decorator)

    def _decorate_result_with_job_finalization_detector(self, job: IsFuncArgJob,
                                                        job_result: object):
        def _register_job_finished() -> None:
            self._register_job_state(job, RunState.FINISHED)

        return decorate_result_by_type(on_finished=_register_job_finished)(lambda: job_result)()

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

    def _run_task(self, state: Any, task: TaskRunSpec, *args, **kwargs) -> object:
        raise NotImplementedError

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

    def _run_flow(self, state: Any, flow: FlowRunSpec, *args, **kwargs) -> object:
        raise NotImplementedError

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 

supported_job_types: frozenset[JobType.Literals] = frozenset()

__init__ 

__init__() -> None

Initialize engine config, registry holder, and subclass state.

Source code in src/omnipy/engine/_base.py 
13
14
15
16
17
18
19
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 
 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
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 abstractmethod 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/engine/_base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
@abstractmethod
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/engine/_base.py 
45
46
47
48
49
50
51
52
53
54
55
56
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 
58
59
60
61
62
63
64
65
66
67
68
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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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