omnipy.compute.flow

Flow definitions for composing tasks and subflows.

This module exposes Omnipy's public flow types. Use LinearFlowTemplate for sequential pipelines, DagFlowTemplate for dependency-driven directed acyclic graphs, and FuncFlowTemplate for flows expressed as a single coordinating callable.

ATTRIBUTE	DESCRIPTION
LinearFlowTemplate	Decorator-style template factory for sequential flows.
DagFlowTemplate	Decorator-style template factory for directed acyclic graph flows.
FuncFlowTemplate	Decorator-style template factory for callable-backed coordinating flows.

CLASS	DESCRIPTION
DagFlow
DagFlowTemplateCore
FlowBase	Provide a shared marker base for Omnipy flow objects.
FuncFlow
FuncFlowTemplateCore	Implement the core template behavior for flows.
LinearFlow	Execute a flow whose tasks run in declaration order.
LinearFlowTemplateCore

DagFlowTemplate module-attribute

DagFlowTemplate = dag_flow_template_as_callable_decorator(
    to_dag_flow_template_init_protocol(DagFlowTemplateCore)
)

Decorator-style factory for defining directed acyclic graph flows.

Use @DagFlowTemplate(...) when task dependencies form a DAG with possible branching and joining, but no cycles.

FuncFlowTemplate module-attribute

FuncFlowTemplate = func_flow_template_as_callable_decorator(
    to_func_flow_template_init_protocol(FuncFlowTemplateCore)
)

Decorator-style factory for defining callable-backed coordinating flows.

Use @FuncFlowTemplate() when a Python callable should orchestrate the flow imperatively.

LinearFlowTemplate module-attribute

LinearFlowTemplate = linear_flow_template_as_callable_decorator(
    to_linear_flow_template_init_protocol(LinearFlowTemplateCore)
)

Decorator-style factory for defining sequential flows.

Use @LinearFlowTemplate(...) when a flow should execute tasks in a fixed, step-by-step order.

DagFlow

Bases: JobMixin[IsDagFlowTemplate[_CallP, _RetT], IsDagFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, ChildJobListArgJobBase[IsDagFlowTemplate[_CallP, _RetT], IsDagFlow[_CallP, _RetT], _CallP, _RetT], Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.DagFlow[DagFlow]
              omnipy.compute._job.JobMixin[JobMixin]
              omnipy.compute.flow.FlowBase[FlowBase]
              omnipy.compute._joblist_job.ChildJobListArgJobBase[ChildJobListArgJobBase]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]

                              omnipy.compute._job.JobMixin --> omnipy.compute.flow.DagFlow
                                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobMixin
                

                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.DagFlow
                
                omnipy.compute._joblist_job.ChildJobListArgJobBase --> omnipy.compute.flow.DagFlow
                                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute._joblist_job.ChildJobListArgJobBase
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                






              click omnipy.compute.flow.DagFlow href "" "omnipy.compute.flow.DagFlow"
              click omnipy.compute._job.JobMixin href "" "omnipy.compute._job.JobMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
              click omnipy.compute._joblist_job.ChildJobListArgJobBase href "" "omnipy.compute._joblist_job.ChildJobListArgJobBase"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
            

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
create_job	Create an applied job instance from the concrete job class.
log	Emit a log message, optionally using an explicit event timestamp.
reset_mixins	Clear all accepted mixins and restore the original init signature.
revise	Return a template reconstructed from this applied job.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
child_job_templates	TYPE: tuple[IsFuncArgJobTemplate, ...]
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger
time_of_cur_toplevel_flow_run	Return the start time of the active top-level flow run, if any. TYPE: datetime | None

Source code in src/omnipy/compute/flow.py

class DagFlow(
        JobMixin[IsDagFlowTemplate[_CallP, _RetT], IsDagFlow[_CallP, _RetT], _CallP, _RetT],
        FlowBase,
        ChildJobListArgJobBase[
            IsDagFlowTemplate[_CallP, _RetT],
            IsDagFlow[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        Generic[_CallP, _RetT],
):
    def _apply_engine_decorator(self, engine: IsEngine) -> None:
        """Register the engine decorator for DAG-flow execution.

        When a runner engine is bound to the flow, this method asks that
        engine to wrap the flow's call function with DAG-specific execution
        behavior.

        Args:
            self: Current DAG flow instance.
            engine: Engine candidate supplied during job setup.
        """
        if self.engine:
            engine = cast(IsJobRunnerEngine, self.engine)
            self_with_mixins = cast(IsDagFlow, self)
            engine.apply_job_decorator(
                JobType.DAG_FLOW,
                self_with_mixins,
                self._accept_call_func_decorator,
            )

    @classmethod
    def _get_job_template_subcls_for_revise(cls) -> type[IsDagFlowTemplate[_CallP, _RetT]]:
        """Return the template type used to revise this flow.

        Revision operations use this hook to recover the decorator-backed DAG
        flow template class corresponding to an executable DAG flow instance.

        Returns:
            type[IsDagFlowTemplate[_CallP, _RetT]]: The ``DagFlowTemplate``
                class associated with this flow.
        """
        return cast(type[IsDagFlowTemplate[_CallP, _RetT]], DagFlowTemplate)

callable_type property

callable_type: CallableType.Literals

child_job_templates property

child_job_templates: tuple[IsFuncArgJobTemplate, ...]

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

time_of_cur_toplevel_flow_run property

time_of_cur_toplevel_flow_run: datetime | None

Return the start time of the active top-level flow run, if any.

RETURNS	DESCRIPTION
datetime | None	datetime | None: Timestamp for the current outermost flow run, or None.

__init__

__init__(*args, **kwargs)

Source code in src/omnipy/compute/_job.py

def __init__(self, *args, **kwargs):
    if JobBase not in self.__class__.__mro__:
        raise TypeError('JobMixin is not meant to be instantiated outside the context '
                        'of a JobBase subclass.')

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

create_job classmethod

create_job(*args: object, **kwargs: object) -> _JobT

Create an applied job instance from the concrete job class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobT	New applied job instance. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job(cls, *args: object, **kwargs: object) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_CREATE_JOB_SUMMARY}}
    #
    # {{ISJOB_CREATE_JOB_DETAILS}}
    """Create an applied job instance from the concrete job class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobT: New applied job instance.
    """
    cls_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], cls)
    return cls_as_job_base._create_job(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

revise

revise() -> _JobTemplateT

Return a template reconstructed from this applied job.

RETURNS	DESCRIPTION
_JobTemplateT	Template carrying the current job configuration. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

def revise(self) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_REVISE_SUMMARY}}
    #
    # {{ISJOB_REVISE_DETAILS}}
    """Return a template reconstructed from this applied job.

    Returns:
        _JobTemplateT: Template carrying the current job configuration.
    """
    self_as_job_base = cast(
        IsJobBase[IsJobTemplate[_JobTemplateT, _JobT, _CallP, _RetT], _JobT, _CallP, _RetT],
        self)
    job_template = self_as_job_base._revise()
    update_wrapper(job_template, self, updated=[])
    return cast(_JobTemplateT, job_template)

DagFlowTemplateCore

Bases: ChildJobListArgJobBase[IsDagFlowTemplate[_CallP, _RetT], IsDagFlow[_CallP, _RetT], _CallP, _RetT], JobTemplateMixin[IsDagFlowTemplate[_CallP, _RetT], IsDagFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.DagFlowTemplateCore[DagFlowTemplateCore]
              omnipy.compute._joblist_job.ChildJobListArgJobBase[ChildJobListArgJobBase]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]
              omnipy.compute._job.JobTemplateMixin[JobTemplateMixin]
              omnipy.compute.flow.FlowBase[FlowBase]

                              omnipy.compute._joblist_job.ChildJobListArgJobBase --> omnipy.compute.flow.DagFlowTemplateCore
                                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute._joblist_job.ChildJobListArgJobBase
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                




                omnipy.compute._job.JobTemplateMixin --> omnipy.compute.flow.DagFlowTemplateCore
                
                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.DagFlowTemplateCore
                


              click omnipy.compute.flow.DagFlowTemplateCore href "" "omnipy.compute.flow.DagFlowTemplateCore"
              click omnipy.compute._joblist_job.ChildJobListArgJobBase href "" "omnipy.compute._joblist_job.ChildJobListArgJobBase"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
              click omnipy.compute._job.JobTemplateMixin href "" "omnipy.compute._job.JobTemplateMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
            

Used by:

• Reference Code reference
  •  omnipy  DagFlowTemplate
  • omnipy compute  flow
    •  DagFlowTemplate
    •  to_dag_flow_template_init_protocol

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
apply	Create an applied job from this template without executing it.
create_job_template	Create a job template instance from the concrete template class.
log	Emit a log message, optionally using an explicit event timestamp.
refine	Forward refinement to the shared template lifecycle implementation.
reset_mixins	Clear all accepted mixins and restore the original init signature.
run	Apply the template and execute the resulting job immediately.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
child_job_templates	TYPE: tuple[IsFuncArgJobTemplate, ...]
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger

Source code in src/omnipy/compute/flow.py

class DagFlowTemplateCore(ChildJobListArgJobBase[IsDagFlowTemplate[_CallP, _RetT],
                                                 IsDagFlow[_CallP, _RetT],
                                                 _CallP,
                                                 _RetT],
                          JobTemplateMixin[IsDagFlowTemplate[_CallP, _RetT],
                                           IsDagFlow[_CallP, _RetT],
                                           _CallP,
                                           _RetT],
                          FlowBase,
                          Generic[_CallP, _RetT]):
    @classmethod
    def _get_job_subcls_for_apply(cls) -> type[IsDagFlow[_CallP, _RetT]]:
        """Return the executable DAG flow type produced by this template.

        The template/application machinery calls this hook when it needs the
        concrete flow class to instantiate from a DAG flow template.

        Returns:
            type[IsDagFlow[_CallP, _RetT]]: The executable ``DagFlow`` subclass
                associated with this template.
        """
        return cast(type[IsDagFlow[_CallP, _RetT]], DagFlow[_CallP, _RetT])

callable_type property

callable_type: CallableType.Literals

child_job_templates property

child_job_templates: tuple[IsFuncArgJobTemplate, ...]

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

__init__

__init__(
    job_func: Callable[_CallP, _RetT],
    /,
    *child_job_templates: IsFuncArgJobTemplate,
    **kwargs: object,
) -> None

Source code in src/omnipy/compute/_joblist_job.py

def __init__(self,
             job_func: Callable[_CallP, _RetT],
             /,
             *child_job_templates: IsFuncArgJobTemplate,
             **kwargs: object) -> None:
    super().__init__(job_func, *child_job_templates, **kwargs)
    self._child_job_templates: tuple[IsFuncArgJobTemplate, ...] = child_job_templates
    self._validate_callable_type_against_child_job_templates()

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

apply

apply() -> _JobT

Create an applied job from this template without executing it.

RETURNS	DESCRIPTION
_JobT	Applied job instance ready to be called. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

def apply(self) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_APPLY_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_APPLY_DETAILS}}
    """Create an applied job from this template without executing it.

    Returns:
        _JobT: Applied job instance ready to be called.
    """
    job = self._cast_to_job_tmpl()._apply()
    update_wrapper(job, self, updated=[])
    return cast(_JobT, job)

create_job_template classmethod

create_job_template(*args: object, **kwargs: object) -> _JobTemplateT

Create a job template instance from the concrete template class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobTemplateT	New job template instance. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job_template(cls, *args: object, **kwargs: object) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_DETAILS}}
    """Create a job template instance from the concrete template class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobTemplateT: New job template instance.
    """

    cls_as_job_base = cast(type[IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT]], cls)
    return cls_as_job_base._create_job_template(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

refine

refine(*args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT

Forward refinement to the shared template lifecycle implementation.

See IsFuncArgJobTemplate.refine and IsChildJobListArgJobTemplate.refine.

Source code in src/omnipy/compute/_job.py

def refine(self, *args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT:
    """Forward refinement to the shared template lifecycle implementation.

    See [`IsFuncArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobTemplate.refine] and
    [`IsChildJobListArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsChildJobListArgJobTemplate.refine].
    """
    self_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], self)
    return self_as_job_base._refine(*args, update=update, **kwargs)

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

run

run(*args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT

Apply the template and execute the resulting job immediately.

PARAMETER	DESCRIPTION
*args	Positional arguments passed to the applied job. TYPE: _CallP.args DEFAULT: ()
**kwargs	Keyword arguments passed to the applied job. TYPE: _CallP.kwargs DEFAULT: {}

RETURNS	DESCRIPTION
_RetCovT	Result returned by the applied job. TYPE: _RetT

Source code in src/omnipy/compute/_job.py

def run(self, *args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_RUN_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_RUN_DETAILS}}
    """Apply the template and execute the resulting job immediately.

    Args:
        *args: Positional arguments passed to the applied job.
        **kwargs: Keyword arguments passed to the applied job.

    Returns:
        _RetCovT: Result returned by the applied job.
    """
    # TODO: Using JobTemplateMixin.run() inside flows should give error message
    return self._cast_to_job_tmpl().apply()(*args, **kwargs)

FlowBase

Provide a shared marker base for Omnipy flow objects.

FlowBase exists to give concrete flow templates and executable flow instances a common nominal base type. It does not define behavior on its own, but it makes flow-specific mixin registration and type-based checks possible within the compute subsystem.

Subclassed by:

• Reference Code reference
  •  omnipy
    •  DagFlow
    •  FuncFlow
    •  LinearFlow
  • omnipy compute  flow
    •  DagFlow
    •  DagFlowTemplateCore
    •  FuncFlow
    •  FuncFlowTemplateCore
    •  LinearFlow
    •  LinearFlowTemplateCore

Source code in src/omnipy/compute/flow.py

class FlowBase:
    """Provide a shared marker base for Omnipy flow objects.

    ``FlowBase`` exists to give concrete flow templates and executable flow
    instances a common nominal base type. It does not define behavior on its
    own, but it makes flow-specific mixin registration and type-based checks
    possible within the compute subsystem.
    """

    ...

FuncFlow

Bases: JobMixin[IsFuncFlowTemplate[_CallP, _RetT], IsFuncFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, FuncArgJobBase[IsFuncFlowTemplate[_CallP, _RetT], IsFuncFlow[_CallP, _RetT], _CallP, _RetT], Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.FuncFlow[FuncFlow]
              omnipy.compute._job.JobMixin[JobMixin]
              omnipy.compute.flow.FlowBase[FlowBase]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]

                              omnipy.compute._job.JobMixin --> omnipy.compute.flow.FuncFlow
                                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobMixin
                

                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.FuncFlow
                
                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute.flow.FuncFlow
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                





              click omnipy.compute.flow.FuncFlow href "" "omnipy.compute.flow.FuncFlow"
              click omnipy.compute._job.JobMixin href "" "omnipy.compute._job.JobMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
            

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
create_job	Create an applied job instance from the concrete job class.
log	Emit a log message, optionally using an explicit event timestamp.
reset_mixins	Clear all accepted mixins and restore the original init signature.
revise	Return a template reconstructed from this applied job.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger
time_of_cur_toplevel_flow_run	Return the start time of the active top-level flow run, if any. TYPE: datetime | None

Source code in src/omnipy/compute/flow.py

class FuncFlow(
        JobMixin[
            IsFuncFlowTemplate[_CallP, _RetT],
            IsFuncFlow[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        FlowBase,
        FuncArgJobBase[
            IsFuncFlowTemplate[_CallP, _RetT],
            IsFuncFlow[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        Generic[_CallP, _RetT],
):
    def _apply_engine_decorator(self, engine: IsEngine) -> None:
        """Register the engine decorator for callable-backed flow execution.

        When a runner engine is bound to the flow, this method asks that
        engine to wrap the coordinating callable with function-flow execution
        behavior.

        Args:
            self: Current function flow instance.
            engine: Engine candidate supplied during job setup.
        """
        if self.engine:
            engine = cast(IsJobRunnerEngine, self.engine)
            self_with_mixins = cast(IsFuncFlow, self)
            engine.apply_job_decorator(
                JobType.FUNC_FLOW,
                self_with_mixins,
                self._accept_call_func_decorator,
            )

    @classmethod
    def _get_job_template_subcls_for_revise(cls) -> type[IsFuncFlowTemplate[_CallP, _RetT]]:
        """Return the template type used to revise this flow.

        Revision operations use this hook to recover the decorator-backed
        function flow template class corresponding to an executable function
        flow instance.

        Returns:
            type[IsFuncFlowTemplate[_CallP, _RetT]]: The ``FuncFlowTemplate``
                class associated with this flow.
        """
        return cast(type[IsFuncFlowTemplate[_CallP, _RetT]], FuncFlowTemplate)

callable_type property

callable_type: CallableType.Literals

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

time_of_cur_toplevel_flow_run property

time_of_cur_toplevel_flow_run: datetime | None

Return the start time of the active top-level flow run, if any.

RETURNS	DESCRIPTION
datetime | None	datetime | None: Timestamp for the current outermost flow run, or None.

__init__

__init__(*args, **kwargs)

Source code in src/omnipy/compute/_job.py

def __init__(self, *args, **kwargs):
    if JobBase not in self.__class__.__mro__:
        raise TypeError('JobMixin is not meant to be instantiated outside the context '
                        'of a JobBase subclass.')

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

create_job classmethod

create_job(*args: object, **kwargs: object) -> _JobT

Create an applied job instance from the concrete job class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobT	New applied job instance. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job(cls, *args: object, **kwargs: object) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_CREATE_JOB_SUMMARY}}
    #
    # {{ISJOB_CREATE_JOB_DETAILS}}
    """Create an applied job instance from the concrete job class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobT: New applied job instance.
    """
    cls_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], cls)
    return cls_as_job_base._create_job(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

revise

revise() -> _JobTemplateT

Return a template reconstructed from this applied job.

RETURNS	DESCRIPTION
_JobTemplateT	Template carrying the current job configuration. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

def revise(self) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_REVISE_SUMMARY}}
    #
    # {{ISJOB_REVISE_DETAILS}}
    """Return a template reconstructed from this applied job.

    Returns:
        _JobTemplateT: Template carrying the current job configuration.
    """
    self_as_job_base = cast(
        IsJobBase[IsJobTemplate[_JobTemplateT, _JobT, _CallP, _RetT], _JobT, _CallP, _RetT],
        self)
    job_template = self_as_job_base._revise()
    update_wrapper(job_template, self, updated=[])
    return cast(_JobTemplateT, job_template)

FuncFlowTemplateCore

Bases: FuncArgJobBase[IsFuncFlowTemplate[_CallP, _RetT], IsFuncFlow[_CallP, _RetT], _CallP, _RetT], JobTemplateMixin[IsFuncFlowTemplate[_CallP, _RetT], IsFuncFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.FuncFlowTemplateCore[FuncFlowTemplateCore]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]
              omnipy.compute._job.JobTemplateMixin[JobTemplateMixin]
              omnipy.compute.flow.FlowBase[FlowBase]

                              omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute.flow.FuncFlowTemplateCore
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                



                omnipy.compute._job.JobTemplateMixin --> omnipy.compute.flow.FuncFlowTemplateCore
                
                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.FuncFlowTemplateCore
                


              click omnipy.compute.flow.FuncFlowTemplateCore href "" "omnipy.compute.flow.FuncFlowTemplateCore"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
              click omnipy.compute._job.JobTemplateMixin href "" "omnipy.compute._job.JobTemplateMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
            

Implement the core template behavior for flows.

A function flow template wraps a Python callable that orchestrates work as a flow. Use this when the control flow is easiest to express directly in Python instead of as an explicit task list or dependency graph.

Instances are normally produced through the FuncFlowTemplate decorator factory rather than by direct construction.

Used by:

• Reference Code reference
  •  omnipy  FuncFlowTemplate
  • omnipy compute  flow
    •  FuncFlowTemplate
    •  to_func_flow_template_init_protocol

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
apply	Create an applied job from this template without executing it.
create_job_template	Create a job template instance from the concrete template class.
log	Emit a log message, optionally using an explicit event timestamp.
refine	Forward refinement to the shared template lifecycle implementation.
reset_mixins	Clear all accepted mixins and restore the original init signature.
run	Apply the template and execute the resulting job immediately.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger

Source code in src/omnipy/compute/flow.py

class FuncFlowTemplateCore(FuncArgJobBase[IsFuncFlowTemplate[_CallP, _RetT],
                                          IsFuncFlow[_CallP, _RetT],
                                          _CallP,
                                          _RetT],
                           JobTemplateMixin[IsFuncFlowTemplate[_CallP, _RetT],
                                            IsFuncFlow[_CallP, _RetT],
                                            _CallP,
                                            _RetT],
                           FlowBase,
                           Generic[_CallP, _RetT]):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Implement the core template behavior for flows.
    #
    # A function flow template wraps a Python callable that orchestrates work as
    # a flow. Use this when the control flow is easiest to express directly in
    # Python instead of as an explicit task list or dependency graph.
    #
    # Instances are normally produced through the ``FuncFlowTemplate`` decorator
    # factory rather than by direct construction.
    #
    """Implement the core template behavior for flows.

    A function flow template wraps a Python callable that orchestrates work as
    a flow. Use this when the control flow is easiest to express directly in
    Python instead of as an explicit task list or dependency graph.

    Instances are normally produced through the ``FuncFlowTemplate`` decorator
    factory rather than by direct construction.
    """
    @classmethod
    def _get_job_subcls_for_apply(cls) -> type[IsFuncFlow[_CallP, _RetT]]:
        """Return the executable function flow type produced by this template.

        The template/application machinery calls this hook when it needs the
        concrete flow class that should be instantiated from a function flow
        template.

        Returns:
            type[IsFuncFlow[_CallP, _RetT]]: The executable ``FuncFlow``
                subclass associated with this template.
        """
        return cast(type[IsFuncFlow[_CallP, _RetT]], FuncFlow[_CallP, _RetT])

callable_type property

callable_type: CallableType.Literals

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

__init__

__init__(job_func: Callable[_CallP, _RetT], /, *args: object, **kwargs: object) -> None

Source code in src/omnipy/compute/_func_job.py

def __init__(self, job_func: Callable[_CallP, _RetT], /, *args: object,
             **kwargs: object) -> None:
    self._job_func = job_func

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

apply

apply() -> _JobT

Create an applied job from this template without executing it.

RETURNS	DESCRIPTION
_JobT	Applied job instance ready to be called. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

def apply(self) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_APPLY_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_APPLY_DETAILS}}
    """Create an applied job from this template without executing it.

    Returns:
        _JobT: Applied job instance ready to be called.
    """
    job = self._cast_to_job_tmpl()._apply()
    update_wrapper(job, self, updated=[])
    return cast(_JobT, job)

create_job_template classmethod

create_job_template(*args: object, **kwargs: object) -> _JobTemplateT

Create a job template instance from the concrete template class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobTemplateT	New job template instance. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job_template(cls, *args: object, **kwargs: object) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_DETAILS}}
    """Create a job template instance from the concrete template class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobTemplateT: New job template instance.
    """

    cls_as_job_base = cast(type[IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT]], cls)
    return cls_as_job_base._create_job_template(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

refine

refine(*args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT

Forward refinement to the shared template lifecycle implementation.

See IsFuncArgJobTemplate.refine and IsChildJobListArgJobTemplate.refine.

Source code in src/omnipy/compute/_job.py

def refine(self, *args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT:
    """Forward refinement to the shared template lifecycle implementation.

    See [`IsFuncArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobTemplate.refine] and
    [`IsChildJobListArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsChildJobListArgJobTemplate.refine].
    """
    self_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], self)
    return self_as_job_base._refine(*args, update=update, **kwargs)

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

run

run(*args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT

Apply the template and execute the resulting job immediately.

PARAMETER	DESCRIPTION
*args	Positional arguments passed to the applied job. TYPE: _CallP.args DEFAULT: ()
**kwargs	Keyword arguments passed to the applied job. TYPE: _CallP.kwargs DEFAULT: {}

RETURNS	DESCRIPTION
_RetCovT	Result returned by the applied job. TYPE: _RetT

Source code in src/omnipy/compute/_job.py

def run(self, *args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_RUN_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_RUN_DETAILS}}
    """Apply the template and execute the resulting job immediately.

    Args:
        *args: Positional arguments passed to the applied job.
        **kwargs: Keyword arguments passed to the applied job.

    Returns:
        _RetCovT: Result returned by the applied job.
    """
    # TODO: Using JobTemplateMixin.run() inside flows should give error message
    return self._cast_to_job_tmpl().apply()(*args, **kwargs)

LinearFlow

Bases: JobMixin[IsLinearFlowTemplate[_CallP, _RetT], IsLinearFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, ChildJobListArgJobBase[IsLinearFlowTemplate[_CallP, _RetT], IsLinearFlow[_CallP, _RetT], _CallP, _RetT], Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.LinearFlow[LinearFlow]
              omnipy.compute._job.JobMixin[JobMixin]
              omnipy.compute.flow.FlowBase[FlowBase]
              omnipy.compute._joblist_job.ChildJobListArgJobBase[ChildJobListArgJobBase]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]

                              omnipy.compute._job.JobMixin --> omnipy.compute.flow.LinearFlow
                                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobMixin
                

                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.LinearFlow
                
                omnipy.compute._joblist_job.ChildJobListArgJobBase --> omnipy.compute.flow.LinearFlow
                                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute._joblist_job.ChildJobListArgJobBase
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                






              click omnipy.compute.flow.LinearFlow href "" "omnipy.compute.flow.LinearFlow"
              click omnipy.compute._job.JobMixin href "" "omnipy.compute._job.JobMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
              click omnipy.compute._joblist_job.ChildJobListArgJobBase href "" "omnipy.compute._joblist_job.ChildJobListArgJobBase"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
            

Execute a flow whose tasks run in declaration order.

A LinearFlow runs its constituent tasks in declaration order, making each step wait for the previous one to finish. Use it for pipelines where every stage depends on the output or side effects of the stage before it.

Instances are typically produced by calling a LinearFlowTemplate rather than by constructing LinearFlow directly.

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
create_job	Create an applied job instance from the concrete job class.
log	Emit a log message, optionally using an explicit event timestamp.
reset_mixins	Clear all accepted mixins and restore the original init signature.
revise	Return a template reconstructed from this applied job.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
child_job_templates	TYPE: tuple[IsFuncArgJobTemplate, ...]
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger
time_of_cur_toplevel_flow_run	Return the start time of the active top-level flow run, if any. TYPE: datetime | None

Source code in src/omnipy/compute/flow.py

class LinearFlow(JobMixin[IsLinearFlowTemplate[_CallP, _RetT],
                          IsLinearFlow[_CallP, _RetT],
                          _CallP,
                          _RetT],
                 FlowBase,
                 ChildJobListArgJobBase[IsLinearFlowTemplate[_CallP, _RetT],
                                        IsLinearFlow[_CallP, _RetT],
                                        _CallP,
                                        _RetT],
                 Generic[_CallP, _RetT]):
    """Execute a flow whose tasks run in declaration order.

    A ``LinearFlow`` runs its constituent tasks in declaration order, making
    each step wait for the previous one to finish. Use it for pipelines where
    every stage depends on the output or side effects of the stage before it.

    Instances are typically produced by calling a ``LinearFlowTemplate``
    rather than by constructing ``LinearFlow`` directly.
    """
    def _apply_engine_decorator(self, engine: IsEngine) -> None:
        """Register the engine decorator for linear-flow execution.

        When the flow already holds a runner engine, this method asks that
        engine to wrap the flow's call function with the engine-specific
        linear-flow execution behavior.

        Args:
            self: Current linear flow instance.
            engine: Engine candidate supplied during job setup.
        """
        if self.engine:
            engine = cast(IsJobRunnerEngine, self.engine)
            self_with_mixins = cast(IsLinearFlow, self)
            engine.apply_job_decorator(
                JobType.LINEAR_FLOW,
                self_with_mixins,
                self._accept_call_func_decorator,
            )

    @classmethod
    def _get_job_template_subcls_for_revise(cls) -> type[IsLinearFlowTemplate[_CallP, _RetT]]:
        """Return the template type used to revise this flow.

        Revision operations use this hook to recover the decorator-backed
        template class corresponding to an executable linear flow instance.

        Returns:
            type[IsLinearFlowTemplate[_CallP, _RetT]]: The ``LinearFlowTemplate``
                class associated with this flow.
        """
        return cast(type[IsLinearFlowTemplate[_CallP, _RetT]], LinearFlowTemplate)

callable_type property

callable_type: CallableType.Literals

child_job_templates property

child_job_templates: tuple[IsFuncArgJobTemplate, ...]

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

time_of_cur_toplevel_flow_run property

time_of_cur_toplevel_flow_run: datetime | None

Return the start time of the active top-level flow run, if any.

RETURNS	DESCRIPTION
datetime | None	datetime | None: Timestamp for the current outermost flow run, or None.

__init__

__init__(*args, **kwargs)

Source code in src/omnipy/compute/_job.py

def __init__(self, *args, **kwargs):
    if JobBase not in self.__class__.__mro__:
        raise TypeError('JobMixin is not meant to be instantiated outside the context '
                        'of a JobBase subclass.')

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

create_job classmethod

create_job(*args: object, **kwargs: object) -> _JobT

Create an applied job instance from the concrete job class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobT	New applied job instance. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job(cls, *args: object, **kwargs: object) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_CREATE_JOB_SUMMARY}}
    #
    # {{ISJOB_CREATE_JOB_DETAILS}}
    """Create an applied job instance from the concrete job class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobT: New applied job instance.
    """
    cls_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], cls)
    return cls_as_job_base._create_job(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

revise

revise() -> _JobTemplateT

Return a template reconstructed from this applied job.

RETURNS	DESCRIPTION
_JobTemplateT	Template carrying the current job configuration. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

def revise(self) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOB_REVISE_SUMMARY}}
    #
    # {{ISJOB_REVISE_DETAILS}}
    """Return a template reconstructed from this applied job.

    Returns:
        _JobTemplateT: Template carrying the current job configuration.
    """
    self_as_job_base = cast(
        IsJobBase[IsJobTemplate[_JobTemplateT, _JobT, _CallP, _RetT], _JobT, _CallP, _RetT],
        self)
    job_template = self_as_job_base._revise()
    update_wrapper(job_template, self, updated=[])
    return cast(_JobTemplateT, job_template)

LinearFlowTemplateCore

Bases: ChildJobListArgJobBase[IsLinearFlowTemplate[_CallP, _RetT], IsLinearFlow[_CallP, _RetT], _CallP, _RetT], JobTemplateMixin[IsLinearFlowTemplate[_CallP, _RetT], IsLinearFlow[_CallP, _RetT], _CallP, _RetT], FlowBase, Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.flow.LinearFlowTemplateCore[LinearFlowTemplateCore]
              omnipy.compute._joblist_job.ChildJobListArgJobBase[ChildJobListArgJobBase]
              omnipy.compute._func_job.FuncArgJobBase[FuncArgJobBase]
              omnipy.compute._func_job.PlainFuncArgJobBase[PlainFuncArgJobBase]
              omnipy.compute._job.JobBase[JobBase]
              omnipy.hub.log.mixin.LogMixin[LogMixin]
              omnipy.util.mixin.DynamicMixinAcceptor[DynamicMixinAcceptor]
              omnipy.compute._job.JobTemplateMixin[JobTemplateMixin]
              omnipy.compute.flow.FlowBase[FlowBase]

                              omnipy.compute._joblist_job.ChildJobListArgJobBase --> omnipy.compute.flow.LinearFlowTemplateCore
                                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute._joblist_job.ChildJobListArgJobBase
                                omnipy.compute._func_job.PlainFuncArgJobBase --> omnipy.compute._func_job.FuncArgJobBase
                                omnipy.compute._job.JobBase --> omnipy.compute._func_job.PlainFuncArgJobBase
                                omnipy.hub.log.mixin.LogMixin --> omnipy.compute._job.JobBase
                
                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobBase
                




                omnipy.compute._job.JobTemplateMixin --> omnipy.compute.flow.LinearFlowTemplateCore
                
                omnipy.compute.flow.FlowBase --> omnipy.compute.flow.LinearFlowTemplateCore
                


              click omnipy.compute.flow.LinearFlowTemplateCore href "" "omnipy.compute.flow.LinearFlowTemplateCore"
              click omnipy.compute._joblist_job.ChildJobListArgJobBase href "" "omnipy.compute._joblist_job.ChildJobListArgJobBase"
              click omnipy.compute._func_job.FuncArgJobBase href "" "omnipy.compute._func_job.FuncArgJobBase"
              click omnipy.compute._func_job.PlainFuncArgJobBase href "" "omnipy.compute._func_job.PlainFuncArgJobBase"
              click omnipy.compute._job.JobBase href "" "omnipy.compute._job.JobBase"
              click omnipy.hub.log.mixin.LogMixin href "" "omnipy.hub.log.mixin.LogMixin"
              click omnipy.util.mixin.DynamicMixinAcceptor href "" "omnipy.util.mixin.DynamicMixinAcceptor"
              click omnipy.compute._job.JobTemplateMixin href "" "omnipy.compute._job.JobTemplateMixin"
              click omnipy.compute.flow.FlowBase href "" "omnipy.compute.flow.FlowBase"
            

Used by:

• Reference Code reference
  •  omnipy  LinearFlowTemplate
  • omnipy compute  flow
    •  LinearFlowTemplate
    •  to_linear_flow_template_init_protocol

METHOD	DESCRIPTION
__init__
accept_mixin	Register a mixin class for dynamic composition.
apply	Create an applied job from this template without executing it.
create_job_template	Create a job template instance from the concrete template class.
log	Emit a log message, optionally using an explicit event timestamp.
refine	Forward refinement to the shared template lifecycle implementation.
reset_mixins	Clear all accepted mixins and restore the original init signature.
run	Apply the template and execute the resulting job immediately.

ATTRIBUTE	DESCRIPTION
callable_type	TYPE: CallableType.Literals
child_job_templates	TYPE: tuple[IsFuncArgJobTemplate, ...]
config	Return the job configuration visible to this instance. TYPE: IsJobConfig
engine	Return the engine associated with this job, if any. TYPE: IsEngine | None
in_flow_context	Return whether the job is currently executing inside a flow context. TYPE: bool
logger	Return the logger bound to the concrete instance type. TYPE: Logger

Source code in src/omnipy/compute/flow.py

class LinearFlowTemplateCore(
        ChildJobListArgJobBase[
            IsLinearFlowTemplate[_CallP, _RetT],
            IsLinearFlow[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        JobTemplateMixin[
            IsLinearFlowTemplate[_CallP, _RetT],
            IsLinearFlow[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        FlowBase,
        Generic[_CallP, _RetT],
):
    @classmethod
    def _get_job_subcls_for_apply(cls) -> type[IsLinearFlow[_CallP, _RetT]]:
        """Return the executable flow type produced by this template.

        The template/application machinery calls this hook when it needs the
        concrete job class that should be instantiated from a linear flow
        template.

        Returns:
            type[IsLinearFlow[_CallP, _RetT]]: The executable ``LinearFlow``
                subclass associated with this template.
        """
        return cast(type[IsLinearFlow[_CallP, _RetT]], LinearFlow[_CallP, _RetT])

callable_type property

callable_type: CallableType.Literals

child_job_templates property

child_job_templates: tuple[IsFuncArgJobTemplate, ...]

config property

config: IsJobConfig

Return the job configuration visible to this instance.

RETURNS	DESCRIPTION
IsJobConfig	Active job configuration used for runtime behavior. TYPE: IsJobConfig

engine property

engine: IsEngine | None

Return the engine associated with this job, if any.

RETURNS	DESCRIPTION
IsEngine | None	IsEngine | None: Engine used for decoration and execution, or None.

in_flow_context property

in_flow_context: bool

Return whether the job is currently executing inside a flow context.

RETURNS	DESCRIPTION
bool	True when a surrounding flow context is active. TYPE: bool

logger property

logger: Logger

Return the logger bound to the concrete instance type.

RETURNS	DESCRIPTION
Logger	Logger used by the object for Omnipy log messages. TYPE: Logger

__init__

__init__(
    job_func: Callable[_CallP, _RetT],
    /,
    *child_job_templates: IsFuncArgJobTemplate,
    **kwargs: object,
) -> None

Source code in src/omnipy/compute/_joblist_job.py

def __init__(self,
             job_func: Callable[_CallP, _RetT],
             /,
             *child_job_templates: IsFuncArgJobTemplate,
             **kwargs: object) -> None:
    super().__init__(job_func, *child_job_templates, **kwargs)
    self._child_job_templates: tuple[IsFuncArgJobTemplate, ...] = child_job_templates
    self._validate_callable_type_against_child_job_templates()

accept_mixin classmethod

accept_mixin(mixin_cls: Type) -> None

Register a mixin class for dynamic composition.

PARAMETER	DESCRIPTION
mixin_cls	Mixin class whose __init__ keyword-only parameters should be merged into the acceptor signature. TYPE: Type

Source code in src/omnipy/util/mixin.py

@classmethod
def accept_mixin(cls, mixin_cls: Type) -> None:
    """Register a mixin class for dynamic composition.

    Args:
        mixin_cls: Mixin class whose ``__init__`` keyword-only parameters
            should be merged into the acceptor signature.
    """
    cls._accept_mixin(mixin_cls, update=True)

apply

apply() -> _JobT

Create an applied job from this template without executing it.

RETURNS	DESCRIPTION
_JobT	Applied job instance ready to be called. TYPE: _JobT

Source code in src/omnipy/compute/_job.py

def apply(self) -> _JobT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_APPLY_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_APPLY_DETAILS}}
    """Create an applied job from this template without executing it.

    Returns:
        _JobT: Applied job instance ready to be called.
    """
    job = self._cast_to_job_tmpl()._apply()
    update_wrapper(job, self, updated=[])
    return cast(_JobT, job)

create_job_template classmethod

create_job_template(*args: object, **kwargs: object) -> _JobTemplateT

Create a job template instance from the concrete template class.

PARAMETER	DESCRIPTION
*args	Positional constructor arguments. TYPE: object DEFAULT: ()
**kwargs	Keyword constructor arguments. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_JobTemplateT	New job template instance. TYPE: _JobTemplateT

Source code in src/omnipy/compute/_job.py

@classmethod
def create_job_template(cls, *args: object, **kwargs: object) -> _JobTemplateT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_CREATE_JOB_TEMPLATE_DETAILS}}
    """Create a job template instance from the concrete template class.

    Args:
        *args: Positional constructor arguments.
        **kwargs: Keyword constructor arguments.

    Returns:
        _JobTemplateT: New job template instance.
    """

    cls_as_job_base = cast(type[IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT]], cls)
    return cls_as_job_base._create_job_template(*args, **kwargs)

log

log(log_msg: str, level: int = INFO, datetime_obj: datetime | None = None)

Emit a log message, optionally using an explicit event timestamp.

PARAMETER	DESCRIPTION
log_msg	Message text to send to the logger. TYPE: str
level	Standard library logging level. TYPE: int DEFAULT: INFO
datetime_obj	Timestamp to attach to the record instead of wall-clock time. TYPE: datetime | None DEFAULT: None

Source code in src/omnipy/hub/log/mixin.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{CANLOG_LOG_SUMMARY}}
    #
    # {{CANLOG_LOG_DETAILS}}
    """Emit a log message, optionally using an explicit event timestamp.

    Args:
        log_msg: Message text to send to the logger.
        level: Standard library logging level.
        datetime_obj: Timestamp to attach to the record instead of wall-clock time.
    """
    if self._logger is not None:
        create_time = datetime_obj.timestamp() if datetime_obj else time.time()
        self._logger.log(level, log_msg, extra=dict(timestamp=create_time))

refine

refine(*args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT

Forward refinement to the shared template lifecycle implementation.

See IsFuncArgJobTemplate.refine and IsChildJobListArgJobTemplate.refine.

Source code in src/omnipy/compute/_job.py

def refine(self, *args: Any, update: bool = True, **kwargs: object) -> _JobTemplateT:
    """Forward refinement to the shared template lifecycle implementation.

    See [`IsFuncArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobTemplate.refine] and
    [`IsChildJobListArgJobTemplate.refine`]
    [omnipy.shared.protocols.compute.job.IsChildJobListArgJobTemplate.refine].
    """
    self_as_job_base = cast(IsJobBase[_JobTemplateT, _JobT, _CallP, _RetT], self)
    return self_as_job_base._refine(*args, update=update, **kwargs)

reset_mixins classmethod

reset_mixins()

Clear all accepted mixins and restore the original init signature.

Source code in src/omnipy/util/mixin.py

@classmethod
def reset_mixins(cls):
    """Clear all accepted mixins and restore the original init signature."""
    cls._mixin_classes.clear()
    cls._init_params_per_mixin_cls.clear()
    cls.__init__.__signature__ = cls._orig_init_signature

run

run(*args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT

Apply the template and execute the resulting job immediately.

PARAMETER	DESCRIPTION
*args	Positional arguments passed to the applied job. TYPE: _CallP.args DEFAULT: ()
**kwargs	Keyword arguments passed to the applied job. TYPE: _CallP.kwargs DEFAULT: {}

RETURNS	DESCRIPTION
_RetCovT	Result returned by the applied job. TYPE: _RetT

Source code in src/omnipy/compute/_job.py

def run(self, *args: _CallP.args, **kwargs: _CallP.kwargs) -> _RetT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISJOBTEMPLATE_RUN_SUMMARY}}
    #
    # {{ISJOBTEMPLATE_RUN_DETAILS}}
    """Apply the template and execute the resulting job immediately.

    Args:
        *args: Positional arguments passed to the applied job.
        **kwargs: Keyword arguments passed to the applied job.

    Returns:
        _RetCovT: Result returned by the applied job.
    """
    # TODO: Using JobTemplateMixin.run() inside flows should give error message
    return self._cast_to_job_tmpl().apply()(*args, **kwargs)

dag_flow_template_as_callable_decorator

dag_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsDagFlowTemplate],
) -> Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsDagFlowTemplate[_CallP, _RetT]]]

Wrap a template initializer as a callable decorator factory.

PARAMETER	DESCRIPTION
decorated_cls	DAG-flow template initializer to adapt. TYPE: Callable[Concatenate[_CallableT, _InitP], IsDagFlowTemplate]

RETURNS	DESCRIPTION
Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsDagFlowTemplate[_CallP, _RetT]]]	A decorator factory that converts a Python callable into a DAG flow template.

Used by:

• Reference Code reference
  •  omnipy  DagFlowTemplate
  • omnipy compute  flow  DagFlowTemplate

Source code in src/omnipy/compute/flow.py

def dag_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsDagFlowTemplate]) -> \
        Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsDagFlowTemplate[_CallP, _RetT]]]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{FLOW_WRAP_INITIALIZER_DECORATOR_SUMMARY}}
    #
    # Args:
    #     decorated_cls: DAG-flow template initializer to adapt.
    #
    # Returns:
    #     A decorator factory that converts a Python callable into a DAG flow template.
    #
    """Wrap a template initializer as a callable decorator factory.

    Args:
        decorated_cls: DAG-flow template initializer to adapt.

    Returns:
        A decorator factory that converts a Python callable into a DAG flow template.
    """
    return callable_decorator_cls(
        cast(
            Callable[Concatenate[Callable[_CallP, _RetT], _InitP], IsDagFlowTemplate[_CallP,
                                                                                     _RetT]],
            decorated_cls))

func_flow_template_as_callable_decorator

func_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsFuncFlowTemplate],
) -> Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsFuncFlowTemplate[_CallP, _RetT]]]

Wrap a template initializer as a callable decorator factory.

PARAMETER	DESCRIPTION
decorated_cls	Function-flow template initializer to adapt. TYPE: Callable[Concatenate[_CallableT, _InitP], IsFuncFlowTemplate]

RETURNS	DESCRIPTION
Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsFuncFlowTemplate[_CallP, _RetT]]]	A decorator factory that converts a Python callable into a function flow template.

Used by:

• Reference Code reference
  •  omnipy  FuncFlowTemplate
  • omnipy compute  flow  FuncFlowTemplate

Source code in src/omnipy/compute/flow.py

def func_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsFuncFlowTemplate]) -> \
        Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsFuncFlowTemplate[_CallP, _RetT]]]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{FLOW_WRAP_INITIALIZER_DECORATOR_SUMMARY}}
    #
    # Args:
    #     decorated_cls: Function-flow template initializer to adapt.
    #
    # Returns:
    #     A decorator factory that converts a Python callable into a function flow template.
    #
    """Wrap a template initializer as a callable decorator factory.

    Args:
        decorated_cls: Function-flow template initializer to adapt.

    Returns:
        A decorator factory that converts a Python callable into a function flow template.
    """
    return callable_decorator_cls(
        cast(
            Callable[Concatenate[Callable[_CallP, _RetT], _InitP],
                     IsFuncFlowTemplate[_CallP, _RetT]],
            decorated_cls))

linear_flow_template_as_callable_decorator

linear_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsLinearFlowTemplate],
) -> Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsLinearFlowTemplate[_CallP, _RetT]]]

Wrap a template initializer as a callable decorator factory.

PARAMETER	DESCRIPTION
decorated_cls	Linear-flow template initializer to adapt. TYPE: Callable[Concatenate[_CallableT, _InitP], IsLinearFlowTemplate]

RETURNS	DESCRIPTION
Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsLinearFlowTemplate[_CallP, _RetT]]]	A decorator factory that converts a Python callable into a linear flow template.

Used by:

• Reference Code reference
  •  omnipy  LinearFlowTemplate
  • omnipy compute  flow  LinearFlowTemplate

Source code in src/omnipy/compute/flow.py

def linear_flow_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsLinearFlowTemplate]) -> \
        Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsLinearFlowTemplate[_CallP, _RetT]]]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{FLOW_WRAP_INITIALIZER_DECORATOR_SUMMARY}}
    #
    # Args:
    #     decorated_cls: Linear-flow template initializer to adapt.
    #
    # Returns:
    #     A decorator factory that converts a Python callable into a linear flow template.
    #
    """Wrap a template initializer as a callable decorator factory.

    Args:
        decorated_cls: Linear-flow template initializer to adapt.

    Returns:
        A decorator factory that converts a Python callable into a linear flow template.
    """
    return callable_decorator_cls(
        cast(
            Callable[Concatenate[Callable[_CallP, _RetT], _InitP],
                     IsLinearFlowTemplate[_CallP, _RetT]],
            decorated_cls))

to_dag_flow_template_init_protocol

to_dag_flow_template_init_protocol(
    decorated_cls: Callable[
        Concatenate[Callable[_CallP, _RetT], _InitP], DagFlowTemplateCore[_CallP, _RetT]
    ],
) -> HasChildJobListArgJobTemplateInit[IsDagFlowTemplate[_CallP, _RetT], _CallP, _RetT]

Cast a DAG-flow template initializer to the shared init protocol.

PARAMETER	DESCRIPTION
decorated_cls	DAG-flow template initializer to cast. TYPE: Callable[Concatenate[Callable[_CallP, _RetT], _InitP], DagFlowTemplateCore[_CallP, _RetT]]

RETURNS	DESCRIPTION
HasChildJobListArgJobTemplateInit[IsDagFlowTemplate[_CallP, _RetT], _CallP, _RetT]	The initializer typed as HasChildJobListArgJobTemplateInit.

Used by:

• Reference Code reference
  •  omnipy  DagFlowTemplate
  • omnipy compute  flow  DagFlowTemplate

Source code in src/omnipy/compute/flow.py

def to_dag_flow_template_init_protocol(
    decorated_cls: Callable[Concatenate[Callable[_CallP, _RetT], _InitP],
                            DagFlowTemplateCore[_CallP, _RetT]]
) -> HasChildJobListArgJobTemplateInit[IsDagFlowTemplate[_CallP, _RetT], _CallP, _RetT]:
    """Cast a DAG-flow template initializer to the shared init protocol.

    Args:
        decorated_cls: DAG-flow template initializer to cast.

    Returns:
        The initializer typed as ``HasChildJobListArgJobTemplateInit``.
    """
    return cast(HasChildJobListArgJobTemplateInit[IsDagFlowTemplate[_CallP, _RetT], _CallP, _RetT],
                decorated_cls)

to_func_flow_template_init_protocol

to_func_flow_template_init_protocol(
    decorated_cls: Callable[
        Concatenate[Callable[_CallP, _RetT], _InitP], FuncFlowTemplateCore[_CallP, _RetT]
    ],
) -> HasFuncArgJobTemplateInit[IsFuncFlowTemplate[_CallP, _RetT], _CallP, _RetT]

Cast a template initializer to the shared init protocol.

PARAMETER	DESCRIPTION
decorated_cls	Function-flow template initializer to cast. TYPE: Callable[Concatenate[Callable[_CallP, _RetT], _InitP], FuncFlowTemplateCore[_CallP, _RetT]]

RETURNS	DESCRIPTION
HasFuncArgJobTemplateInit[IsFuncFlowTemplate[_CallP, _RetT], _CallP, _RetT]	The initializer typed as HasFuncArgJobTemplateInit.

Used by:

• Reference Code reference
  •  omnipy  FuncFlowTemplate
  • omnipy compute  flow  FuncFlowTemplate

Source code in src/omnipy/compute/flow.py

def to_func_flow_template_init_protocol(
    decorated_cls: Callable[Concatenate[Callable[_CallP, _RetT], _InitP],
                            FuncFlowTemplateCore[_CallP, _RetT]]
) -> HasFuncArgJobTemplateInit[IsFuncFlowTemplate[_CallP, _RetT], _CallP, _RetT]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{FLOW_CAST_INIT_PROTOCOL_SUMMARY}}
    #
    # Args:
    #     decorated_cls: Function-flow template initializer to cast.
    #
    # Returns:
    #     The initializer typed as ``HasFuncArgJobTemplateInit``.
    #
    """Cast a template initializer to the shared init protocol.

    Args:
        decorated_cls: Function-flow template initializer to cast.

    Returns:
        The initializer typed as ``HasFuncArgJobTemplateInit``.
    """
    return cast(HasFuncArgJobTemplateInit[IsFuncFlowTemplate[_CallP, _RetT], _CallP, _RetT],
                decorated_cls)

to_linear_flow_template_init_protocol

to_linear_flow_template_init_protocol(
    decorated_cls: Callable[
        Concatenate[Callable[_CallP, _RetT], _InitP], LinearFlowTemplateCore[_CallP, _RetT]
    ],
) -> HasChildJobListArgJobTemplateInit[IsLinearFlowTemplate[_CallP, _RetT], _CallP, _RetT]

Cast a linear-flow template initializer to the shared init protocol.

PARAMETER	DESCRIPTION
decorated_cls	Linear-flow template initializer to cast. TYPE: Callable[Concatenate[Callable[_CallP, _RetT], _InitP], LinearFlowTemplateCore[_CallP, _RetT]]

RETURNS	DESCRIPTION
HasChildJobListArgJobTemplateInit[IsLinearFlowTemplate[_CallP, _RetT], _CallP, _RetT]	The initializer typed as HasChildJobListArgJobTemplateInit.

Used by:

• Reference Code reference
  •  omnipy  LinearFlowTemplate
  • omnipy compute  flow  LinearFlowTemplate

Source code in src/omnipy/compute/flow.py

def to_linear_flow_template_init_protocol(
    decorated_cls: Callable[Concatenate[Callable[_CallP, _RetT], _InitP],
                            LinearFlowTemplateCore[_CallP, _RetT]]
) -> HasChildJobListArgJobTemplateInit[IsLinearFlowTemplate[_CallP, _RetT], _CallP, _RetT]:
    """Cast a linear-flow template initializer to the shared init protocol.

    Args:
        decorated_cls: Linear-flow template initializer to cast.

    Returns:
        The initializer typed as ``HasChildJobListArgJobTemplateInit``.
    """
    return cast(
        HasChildJobListArgJobTemplateInit[IsLinearFlowTemplate[_CallP, _RetT], _CallP, _RetT],
        decorated_cls)