omnipy.compute.task

Task definitions for callable-backed compute jobs.

This module exposes Omnipy's public task-building API. Use TaskTemplate to define a reusable task from a Python callable, and use Task for the bound executable job instance created from that template.

ATTRIBUTE	DESCRIPTION
TaskTemplate	Decorator-style task template factory for wrapping a callable as a reusable Omnipy task.

CLASS	DESCRIPTION
Task	Execute a single callable-backed Omnipy task.
TaskBase	Provide a shared marker base for Omnipy task objects.
TaskTemplateCore	Implement the core template behavior for tasks.

TaskTemplate module-attribute

TaskTemplate = task_template_as_callable_decorator(to_task_template_init_protocol(TaskTemplateCore))

Decorator-style factory for defining reusable callable-backed tasks.

Use @TaskTemplate() on a Python callable to create a reusable task template. Calling the resulting template produces executable Task instances that run as single Omnipy compute steps.

Task

Bases: JobMixin[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT], TaskBase, FuncArgJobBase[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT], Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.task.Task[Task]
              omnipy.compute._job.JobMixin[JobMixin]
              omnipy.compute.task.TaskBase[TaskBase]
              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.task.Task
                                omnipy.util.mixin.DynamicMixinAcceptor --> omnipy.compute._job.JobMixin
                

                omnipy.compute.task.TaskBase --> omnipy.compute.task.Task
                
                omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute.task.Task
                                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.task.Task href "" "omnipy.compute.task.Task"
              click omnipy.compute._job.JobMixin href "" "omnipy.compute._job.JobMixin"
              click omnipy.compute.task.TaskBase href "" "omnipy.compute.task.TaskBase"
              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 single callable-backed Omnipy task.

A Task is the runnable job object produced from a TaskTemplate. When invoked, it delegates execution of the wrapped callable to the configured task runner engine.

Use this type when one callable should be scheduled, configured, logged, and optionally persisted as a standalone compute step.

Instances are typically produced by calling a TaskTemplate.

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/task.py

class Task(JobMixin[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT],
           TaskBase,
           FuncArgJobBase[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT],
           Generic[_CallP, _RetT]):
    """Execute a single callable-backed Omnipy task.

    A ``Task`` is the runnable job object produced from a ``TaskTemplate``.
    When invoked, it delegates execution of the wrapped callable to the
    configured task runner engine.

    Use this type when one callable should be scheduled, configured, logged,
    and optionally persisted as a standalone compute step.

    Instances are typically produced by calling a ``TaskTemplate``.
    """
    def _apply_engine_decorator(self, engine: IsEngine) -> None:
        """Register the engine decorator for task execution.

        When a runner engine is bound to the task, this method asks that
        engine to wrap the task's callable with task-specific execution
        behavior.

        Args:
            self: Current task instance.
            engine: Engine candidate supplied during job setup.
        """
        if self.engine:
            engine = cast(IsJobRunnerEngine, self.engine)
            self_with_mixins = cast(IsTask[_CallP, _RetT], self)
            engine.apply_job_decorator(
                JobType.TASK,
                self_with_mixins,
                self._accept_call_func_decorator,
            )

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

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

        Returns:
            type[IsTaskTemplate[_CallP, _RetT]]: The ``TaskTemplate`` class
                associated with this task.
        """
        return cast(type[IsTaskTemplate[_CallP, _RetT]], TaskTemplate)

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)

TaskBase

Provide a shared marker base for Omnipy task objects.

TaskBase exists to give concrete task templates and executable task instances a common nominal base type. It does not add runtime behavior on its own, but it supports task-specific typing and internal mixin handling.

Subclassed by:

• Reference Code reference
  •  omnipy  Task
  • omnipy compute  task
    •  Task
    •  TaskTemplateCore

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

class TaskBase:
    """Provide a shared marker base for Omnipy task objects.

    ``TaskBase`` exists to give concrete task templates and executable task
    instances a common nominal base type. It does not add runtime behavior on
    its own, but it supports task-specific typing and internal mixin handling.
    """

    # TODO: Can this and FlowBase be replaced with IsTask/IsFlow, or similar?
    ...

TaskTemplateCore

Bases: FuncArgJobBase[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT], JobTemplateMixin[IsTaskTemplate[_CallP, _RetT], IsTask[_CallP, _RetT], _CallP, _RetT], TaskBase, Generic[_CallP, _RetT]

              flowchart BT
              omnipy.compute.task.TaskTemplateCore[TaskTemplateCore]
              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.task.TaskBase[TaskBase]

                              omnipy.compute._func_job.FuncArgJobBase --> omnipy.compute.task.TaskTemplateCore
                                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.task.TaskTemplateCore
                
                omnipy.compute.task.TaskBase --> omnipy.compute.task.TaskTemplateCore
                


              click omnipy.compute.task.TaskTemplateCore href "" "omnipy.compute.task.TaskTemplateCore"
              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.task.TaskBase href "" "omnipy.compute.task.TaskBase"
            

Implement the core template behavior for tasks.

A task template wraps a Python callable that performs a single unit of work as a task. Use this when the work can be expressed as a self-contained function call.

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

Used by:

• Reference Code reference
  •  omnipy  TaskTemplate
  • omnipy compute  task
    •  TaskTemplate
    •  to_task_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/task.py

class TaskTemplateCore(
        FuncArgJobBase[
            IsTaskTemplate[_CallP, _RetT],
            IsTask[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        JobTemplateMixin[
            IsTaskTemplate[_CallP, _RetT],
            IsTask[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        TaskBase,
        Generic[_CallP, _RetT],
):
    """Implement the core template behavior for tasks.

    A task template wraps a Python callable that performs a single unit of work
    as a task. Use this when the work can be expressed as a self-contained
    function call.

    Instances are normally produced through the ``TaskTemplate`` decorator
    factory rather than by direct construction.
    """
    @classmethod
    def _get_job_subcls_for_apply(cls) -> type[IsTask[_CallP, _RetT]]:
        """Return the executable task 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 task template.

        Returns:
            type[IsTask[_CallP, _RetT]]: The executable ``Task`` subclass
                associated with this template.
        """
        return cast(type[IsTask[_CallP, _RetT]], Task[_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)

task_template_as_callable_decorator

task_template_as_callable_decorator(
    decorated_cls: Callable[Concatenate[_CallableT, _InitP], IsTaskTemplate],
) -> Callable[_InitP, Callable[[Callable[_CallP, _RetT]], IsTaskTemplate[_CallP, _RetT]]]

Wrap a template initializer as a callable decorator factory.

PARAMETER	DESCRIPTION
decorated_cls	Task-template initializer to adapt. TYPE: Callable[Concatenate[_CallableT, _InitP], IsTaskTemplate]

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

Used by:

• Reference Code reference
  •  omnipy  TaskTemplate
  • omnipy compute  task  TaskTemplate

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

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

    Args:
        decorated_cls: Task-template initializer to adapt.

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

to_task_template_init_protocol

to_task_template_init_protocol(
    decorated_cls: Callable[
        Concatenate[Callable[_CallP, _RetT], _InitP], TaskTemplateCore[_CallP, _RetT]
    ],
) -> HasFuncArgJobTemplateInit[IsTaskTemplate[_CallP, _RetT], _CallP, _RetT]

Cast a template initializer to the shared init protocol.

PARAMETER	DESCRIPTION
decorated_cls	Task-template initializer to cast. TYPE: Callable[Concatenate[Callable[_CallP, _RetT], _InitP], TaskTemplateCore[_CallP, _RetT]]

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

Used by:

• Reference Code reference
  •  omnipy  TaskTemplate
  • omnipy compute  task  TaskTemplate

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

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

    Args:
        decorated_cls: Task-template initializer to cast.

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