Skip to content

omnipy.engine.run_spec

CLASS DESCRIPTION
DagFlowRunSpec
FlowRunSpec
FuncFlowRunSpec
JobRunSpec
LinearFlowRunSpec
TaskRunSpec
TaskTemplateArgsFlowRunSpec

DagFlowRunSpec

Bases: TaskTemplateArgsFlowRunSpec

METHOD DESCRIPTION
__init__
create_default_run_callable
get_bound_args
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
flow_context

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

task_templates

TYPE: tuple[IsTaskTemplate, ...]

Source code in src/omnipy/engine/run_spec.py 
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
class DagFlowRunSpec(TaskTemplateArgsFlowRunSpec):
    def _create_default_run_callable(self) -> Callable:  # noqa: C901
        def _inner_run_dag_flow(*args: object, **kwargs: object):
            results = {}
            result = None
            for i, job in enumerate(self.task_templates):
                if i == 0:
                    results = self.get_bound_args(*args, **kwargs).arguments

                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

flow_context property 

flow_context

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

task_templates property 

task_templates: tuple[IsTaskTemplate, ...]

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
65
66
67
68
69
70
71
72
73
def create_default_run_callable(self) -> Callable:
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        callable_type_from_flags(
            has_async=self.has_async_func(),
            has_generator=self.has_generator_func(),
        ),
        context_factory=lambda: self.flow_context,
    )

get_bound_args 

get_bound_args(*args: object, **kwargs: object) -> BoundArguments
Source code in src/omnipy/engine/run_spec.py 
61
62
63
def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

FlowRunSpec

Bases: JobRunSpec, ABC

METHOD DESCRIPTION
__init__
create_default_run_callable
get_bound_args
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
flow_context

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

Source code in src/omnipy/engine/run_spec.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class FlowRunSpec(JobRunSpec, ABC):
    @property
    def flow_context(self):
        flow = cast(IsAnyFlow, self._job)
        return flow.flow_context

    def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
        flow = cast(IsAnyFlow, self._job)
        return flow.get_bound_args(*args, **kwargs)

    def create_default_run_callable(self) -> Callable:
        return decorate_callable_by_type(
            self._create_default_run_callable(),
            callable_type_from_flags(
                has_async=self.has_async_func(),
                has_generator=self.has_generator_func(),
            ),
            context_factory=lambda: self.flow_context,
        )

    @abstractmethod
    def _create_default_run_callable(self) -> Callable:
        ...

flow_context property 

flow_context

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
65
66
67
68
69
70
71
72
73
def create_default_run_callable(self) -> Callable:
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        callable_type_from_flags(
            has_async=self.has_async_func(),
            has_generator=self.has_generator_func(),
        ),
        context_factory=lambda: self.flow_context,
    )

get_bound_args 

get_bound_args(*args: object, **kwargs: object) -> BoundArguments
Source code in src/omnipy/engine/run_spec.py 
61
62
63
def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

FuncFlowRunSpec

Bases: FlowRunSpec

METHOD DESCRIPTION
__init__
create_default_run_callable
get_bound_args
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
flow_context

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

Source code in src/omnipy/engine/run_spec.py 
140
141
142
class FuncFlowRunSpec(FlowRunSpec):
    def _create_default_run_callable(self) -> Callable:
        return self._run_callable

flow_context property 

flow_context

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
65
66
67
68
69
70
71
72
73
def create_default_run_callable(self) -> Callable:
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        callable_type_from_flags(
            has_async=self.has_async_func(),
            has_generator=self.has_generator_func(),
        ),
        context_factory=lambda: self.flow_context,
    )

get_bound_args 

get_bound_args(*args: object, **kwargs: object) -> BoundArguments
Source code in src/omnipy/engine/run_spec.py 
61
62
63
def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

JobRunSpec

Bases: ABC

METHOD DESCRIPTION
__init__
create_default_run_callable
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

Source code in src/omnipy/engine/run_spec.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class JobRunSpec(ABC):
    def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
        self._job = job
        self._run_callable = run_callable

    @property
    def name(self) -> str:
        return self._job.name

    @property
    def param_signatures(self) -> MappingProxyType:
        return self._job.param_signatures

    @property
    def return_type(self) -> type:
        return self._job.return_type

    def has_async_func(self) -> bool:
        return self._job.has_async_func()

    def has_generator_func(self) -> bool:
        return self._job.has_generator_func()

    def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
        self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

    @abstractmethod
    def create_default_run_callable(self) -> Callable:
        ...

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable abstractmethod 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
40
41
42
@abstractmethod
def create_default_run_callable(self) -> Callable:
    ...

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

LinearFlowRunSpec

Bases: TaskTemplateArgsFlowRunSpec

METHOD DESCRIPTION
__init__
create_default_run_callable
get_bound_args
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
flow_context

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

task_templates

TYPE: tuple[IsTaskTemplate, ...]

Source code in src/omnipy/engine/run_spec.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class LinearFlowRunSpec(TaskTemplateArgsFlowRunSpec):
    def _create_default_run_callable(self) -> Callable:
        def _inner_run_linear_flow(*args: object, **kwargs: object):
            result = None
            for i, job in enumerate(self.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

flow_context property 

flow_context

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

task_templates property 

task_templates: tuple[IsTaskTemplate, ...]

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
65
66
67
68
69
70
71
72
73
def create_default_run_callable(self) -> Callable:
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        callable_type_from_flags(
            has_async=self.has_async_func(),
            has_generator=self.has_generator_func(),
        ),
        context_factory=lambda: self.flow_context,
    )

get_bound_args 

get_bound_args(*args: object, **kwargs: object) -> BoundArguments
Source code in src/omnipy/engine/run_spec.py 
61
62
63
def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

TaskRunSpec

Bases: JobRunSpec

METHOD DESCRIPTION
__init__
create_default_run_callable
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
in_flow_context

TYPE: bool

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

Source code in src/omnipy/engine/run_spec.py 
45
46
47
48
49
50
51
52
class TaskRunSpec(JobRunSpec):
    @property
    def in_flow_context(self) -> bool:
        task = cast(IsTask, self._job)
        return task.in_flow_context

    def create_default_run_callable(self) -> Callable:
        return self._run_callable

in_flow_context property 

in_flow_context: bool

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
51
52
def create_default_run_callable(self) -> Callable:
    return self._run_callable

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

TaskTemplateArgsFlowRunSpec

Bases: FlowRunSpec, ABC

METHOD DESCRIPTION
__init__
create_default_run_callable
get_bound_args
has_async_func
has_generator_func
log
ATTRIBUTE DESCRIPTION
flow_context

name

TYPE: str

param_signatures

TYPE: MappingProxyType

return_type

TYPE: type

task_templates

TYPE: tuple[IsTaskTemplate, ...]

Source code in src/omnipy/engine/run_spec.py 
80
81
82
83
84
class TaskTemplateArgsFlowRunSpec(FlowRunSpec, ABC):
    @property
    def task_templates(self) -> tuple[IsTaskTemplate, ...]:
        flow = cast(IsTaskTemplateArgsJob[IsTaskTemplate, Any, Any, Any, Any], self._job)
        return flow.task_templates

flow_context property 

flow_context

name property 

name: str

param_signatures property 

param_signatures: MappingProxyType

return_type property 

return_type: type

task_templates property 

task_templates: tuple[IsTaskTemplate, ...]

__init__ 

__init__(job: IsFuncArgJob, run_callable: Callable) -> None
Source code in src/omnipy/engine/run_spec.py 
15
16
17
def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable 

create_default_run_callable() -> Callable
Source code in src/omnipy/engine/run_spec.py 
65
66
67
68
69
70
71
72
73
def create_default_run_callable(self) -> Callable:
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        callable_type_from_flags(
            has_async=self.has_async_func(),
            has_generator=self.has_generator_func(),
        ),
        context_factory=lambda: self.flow_context,
    )

get_bound_args 

get_bound_args(*args: object, **kwargs: object) -> BoundArguments
Source code in src/omnipy/engine/run_spec.py 
61
62
63
def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

has_async_func 

has_async_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
31
32
def has_async_func(self) -> bool:
    return self._job.has_async_func()

has_generator_func 

has_generator_func() -> bool
Source code in src/omnipy/engine/run_spec.py 
34
35
def has_generator_func(self) -> bool:
    return self._job.has_generator_func()

log 

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None
Source code in src/omnipy/engine/run_spec.py 
37
38
def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)