Module omnipy.engine.local

Overview

View Source

from typing import Any, Callable, Type

from omnipy.api.protocols.public.compute import IsDagFlow, IsFuncFlow, IsLinearFlow, IsTask

from omnipy.api.protocols.public.config import IsLocalRunnerConfig

from omnipy.config.engine import LocalRunnerConfig

from omnipy.engine.job_runner import (DagFlowRunnerEngine,

                                      FuncFlowRunnerEngine,

                                      LinearFlowRunnerEngine,

                                      TaskRunnerEngine)

class LocalRunner(TaskRunnerEngine,

                  LinearFlowRunnerEngine,

                  DagFlowRunnerEngine,

                  FuncFlowRunnerEngine):

    """Local job runner"""

    def _init_engine(self) -> None:

        ...

    def _update_from_config(self) -> None:

        ...

    @classmethod

    def get_config_cls(cls) -> Type[IsLocalRunnerConfig]:

        return LocalRunnerConfig

    def _init_task(self, task: IsTask, call_func: Callable) -> Any:

        ...

    def _run_task(self, state: Any, task: IsTask, call_func: Callable, *args, **kwargs) -> Any:

        return call_func(*args, **kwargs)

    def _init_linear_flow(self, flow: IsLinearFlow) -> Any:

        ...

    def _run_linear_flow(self, state: Any, flow: IsLinearFlow, *args, **kwargs) -> Any:

        return self.default_linear_flow_run_decorator(flow)(*args, **kwargs)

    def _init_dag_flow(self, flow: IsDagFlow) -> Any:

        ...

    def _run_dag_flow(self, state: Any, flow: IsDagFlow, *args, **kwargs) -> Any:

        return self.default_dag_flow_run_decorator(flow)(*args, **kwargs)

    def _init_func_flow(self, func_flow: IsFuncFlow, call_func: Callable) -> object:

        pass

    def _run_func_flow(self,

                       state: Any,

                       func_flow: IsFuncFlow,

                       call_func: Callable,

                       *args,

                       **kwargs) -> Any:

        with func_flow.flow_context:

            return call_func(*args, **kwargs)

Classes

LocalRunner

class LocalRunner(

)

Local job runner

View Source

class LocalRunner(TaskRunnerEngine,

                  LinearFlowRunnerEngine,

                  DagFlowRunnerEngine,

                  FuncFlowRunnerEngine):

    """Local job runner"""

    def _init_engine(self) -> None:

        ...

    def _update_from_config(self) -> None:

        ...

    @classmethod

    def get_config_cls(cls) -> Type[IsLocalRunnerConfig]:

        return LocalRunnerConfig

    def _init_task(self, task: IsTask, call_func: Callable) -> Any:

        ...

    def _run_task(self, state: Any, task: IsTask, call_func: Callable, *args, **kwargs) -> Any:

        return call_func(*args, **kwargs)

    def _init_linear_flow(self, flow: IsLinearFlow) -> Any:

        ...

    def _run_linear_flow(self, state: Any, flow: IsLinearFlow, *args, **kwargs) -> Any:

        return self.default_linear_flow_run_decorator(flow)(*args, **kwargs)

    def _init_dag_flow(self, flow: IsDagFlow) -> Any:

        ...

    def _run_dag_flow(self, state: Any, flow: IsDagFlow, *args, **kwargs) -> Any:

        return self.default_dag_flow_run_decorator(flow)(*args, **kwargs)

    def _init_func_flow(self, func_flow: IsFuncFlow, call_func: Callable) -> object:

        pass

    def _run_func_flow(self,

                       state: Any,

                       func_flow: IsFuncFlow,

                       call_func: Callable,

                       *args,

                       **kwargs) -> Any:

        with func_flow.flow_context:

            return call_func(*args, **kwargs)

---

Static methods

default_dag_flow_run_decorator

def default_dag_flow_run_decorator(
    dag_flow: omnipy.api.protocols.public.compute.IsDagFlow
) -> Any

Parameters:

Name	Type	Description	Default
dag_flow	IsDagFlow

Returns:

Type	Description
Any

View Source

    @staticmethod

    def default_dag_flow_run_decorator(dag_flow: IsDagFlow) -> Any:  # noqa: C901

        def _inner_run_dag_flow(*args: object, **kwargs: object):

            results = {}

            result = None

            with dag_flow.flow_context:

                for i, job in enumerate(dag_flow.task_templates):

                    if i == 0:

                        results = dag_flow.get_call_args(*args, **kwargs)

                    param_keys = set(inspect.signature(job).parameters.keys())

                    # TODO: Refactor to remove dependency

                    #       Also, add test for not allowing override of fixed_params

                    if hasattr(job, 'param_key_map'):

                        for key, val in job.param_key_map.items():

                            if key in param_keys:

                                param_keys.remove(key)

                                param_keys.add(val)

                    if hasattr(job, 'fixed_params'):

                        for key in job.fixed_params.keys():

                            if key in param_keys:

                                param_keys.remove(key)

                    params = {key: val for key, val in results.items() if key in param_keys}

                    result = job(**params)

                    if isinstance(result, dict) and len(result) > 0:

                        results.update(result)

                    else:

                        results[job.name] = result

            return result

        return _inner_run_dag_flow

default_linear_flow_run_decorator

def default_linear_flow_run_decorator(
    linear_flow: omnipy.api.protocols.public.compute.IsLinearFlow
) -> Any

Parameters:

Name	Type	Description	Default
linear_flow	IsLinearFlow

Returns:

Type	Description
Any

View Source

    @staticmethod

    def default_linear_flow_run_decorator(linear_flow: IsLinearFlow) -> Any:

        def _inner_run_linear_flow(*args: object, **kwargs: object):

            result = None

            with linear_flow.flow_context:

                for i, job in enumerate(linear_flow.task_templates):

                    # TODO: Better handling of kwargs

                    if i == 0:

                        result = job(*args, **kwargs)

                    else:

                        result = job(*args)

                    args = (result,)

            return result

        return _inner_run_linear_flow

get_config_cls

def get_config_cls(

) -> Type[omnipy.api.protocols.public.config.IsLocalRunnerConfig]

Specification of config class mapped to an Engine subclass. Must be implemented by all

subclasses of Engine. If no configuration is needed, then the EngineConfig class should be returned.

Returns:

Type	Description
Type[IsLocalRunnerConfig]	Class implementing the IsEngineConfig protocol

View Source

    @classmethod

    def get_config_cls(cls) -> Type[IsLocalRunnerConfig]:

        return LocalRunnerConfig

Methods

apply_dag_flow_decorator

def apply_dag_flow_decorator(
    self,
    dag_flow: omnipy.api.protocols.public.compute.IsDagFlow,
    job_callback_accept_decorator: Callable
) -> None

Parameters:

Name	Type	Description	Default
dag_flow	IsDagFlow
job_callback_accept_decorator	Callable

Returns:

Type	Description
NoneType

View Source

    def apply_dag_flow_decorator(self, dag_flow: IsDagFlow,

                                 job_callback_accept_decorator: Callable) -> None:

        def _dag_flow_decorator(call_func: Callable) -> Callable:

            self._register_job_state(dag_flow, RunState.INITIALIZED)

            state = self._init_dag_flow(dag_flow)

            def _dag_flow_runner_call_func(*args: object, **kwargs: object) -> Any:

                self._register_job_state(dag_flow, RunState.RUNNING)

                flow_result = self._run_dag_flow(state, dag_flow, *args, **kwargs)

                return self._decorate_result_with_job_finalization_detector(dag_flow, flow_result)

            return _dag_flow_runner_call_func

        job_callback_accept_decorator(_dag_flow_decorator)

apply_func_flow_decorator

def apply_func_flow_decorator(
    self,
    func_flow: omnipy.api.protocols.public.compute.IsFuncFlow,
    job_callback_accept_decorator: Callable
) -> None

Parameters:

Name	Type	Description	Default
func_flow	IsFuncFlow
job_callback_accept_decorator	Callable

Returns:

Type	Description
NoneType

View Source

    def apply_func_flow_decorator(self,

                                  func_flow: IsFuncFlow,

                                  job_callback_accept_decorator: Callable) -> None:

        def _func_flow_decorator(call_func: Callable) -> Callable:

            self._register_job_state(func_flow, RunState.INITIALIZED)

            state = self._init_func_flow(func_flow, call_func)

            def _func_flow_runner_call_func(*args: object, **kwargs: object) -> Any:

                self._register_job_state(func_flow, RunState.RUNNING)

                with func_flow.flow_context:

                    flow_result = self._run_func_flow(state, func_flow, call_func, *args, **kwargs)

                    return self._decorate_result_with_job_finalization_detector(

                        func_flow, flow_result)

            return _func_flow_runner_call_func

        job_callback_accept_decorator(_func_flow_decorator)

apply_linear_flow_decorator

def apply_linear_flow_decorator(
    self,
    linear_flow: omnipy.api.protocols.public.compute.IsLinearFlow,
    job_callback_accept_decorator: Callable
) -> None

Parameters:

Name	Type	Description	Default
linear_flow	IsLinearFlow
job_callback_accept_decorator	Callable

Returns:

Type	Description
NoneType

View Source

    def apply_linear_flow_decorator(self,

                                    linear_flow: IsLinearFlow,

                                    job_callback_accept_decorator: Callable) -> None:

        def _linear_flow_decorator(call_func: Callable) -> Callable:

            self._register_job_state(linear_flow, RunState.INITIALIZED)

            state = self._init_linear_flow(linear_flow)

            def _linear_flow_runner_call_func(*args: object, **kwargs: object) -> Any:

                self._register_job_state(linear_flow, RunState.RUNNING)

                flow_result = self._run_linear_flow(state, linear_flow, *args, **kwargs)

                return self._decorate_result_with_job_finalization_detector(

                    linear_flow, flow_result)

            return _linear_flow_runner_call_func

        job_callback_accept_decorator(_linear_flow_decorator)

apply_task_decorator

def apply_task_decorator(
    self,
    task: omnipy.api.protocols.public.compute.IsTask,
    job_callback_accept_decorator: Callable
) -> None

Parameters:

Name	Type	Description	Default
task	IsTask
job_callback_accept_decorator	Callable

Returns:

Type	Description
NoneType

View Source

    def apply_task_decorator(self, task: IsTask, job_callback_accept_decorator: Callable) -> None:

        def _task_decorator(call_func: Callable) -> Callable:

            self._register_job_state(task, RunState.INITIALIZED)

            state = self._init_task(task, call_func)

            def _task_runner_call_func(*args: object, **kwargs: object) -> Any:

                self._register_job_state(task, RunState.RUNNING)

                task_result = self._run_task(state, task, call_func, *args, **kwargs)

                return self._decorate_result_with_job_finalization_detector(task, task_result)

            return _task_runner_call_func

        job_callback_accept_decorator(_task_decorator)

set_config

def set_config(
    self,
    config: omnipy.api.protocols.public.config.IsEngineConfig
) -> None

Parameters:

Name	Type	Description	Default
config	IsEngineConfig

Returns:

Type	Description
NoneType

View Source

    def set_config(self, config: IsEngineConfig) -> None:

        self._config = config

        self._update_from_config()

set_registry

def set_registry(
    self,
    registry: omnipy.api.protocols.private.log.IsRunStateRegistry | None
) -> None

Parameters:

Name	Type	Description	Default
registry	omnipy.api.protocols.private.log.IsRunStateRegistry	None

Returns:

Type	Description
NoneType

View Source

    def set_registry(self, registry: IsRunStateRegistry | None) -> None:

        self._registry = registry