Skip to content

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.

TYPE: Any

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.
FUNCTION DESCRIPTION
TaskTemplate

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

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 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
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]], TaskTemplateCore)

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 
320
321
322
323
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 
 97
 98
 99
100
101
102
103
104
105
@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 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
@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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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 
155
156
157
158
159
160
@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

Referenced by:
Source code in src/omnipy/compute/_job.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
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:
Source code in src/omnipy/compute/task.py 
84
85
86
87
88
89
90
91
92
93
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.

Decorator usage

Apply the template factory as a decorator to a Python callable. The wrapped callable becomes a reusable job template whose public outer signature is visible to template users and to the applied jobs created from it.

Wrapped callable

The wrapped callable defines both the implementation and the public outer signature of the task.

Examples:

>>> import omnipy as om
>>> class TextModel(om.Model[str]):
...     ...

>>> class TextDataset(om.Dataset[TextModel]):
...     ...

>>> @om.TaskTemplate()
... def add_suffix(
...     dataset: TextDataset,
...     suffix: str,
... ) -> TextModel:
...     for data_file_name, value in dataset.items():
...         dataset[data_file_name] = f'{value}{suffix}'
...     return dataset

>>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
>>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
>>> add_suffix.run(text_files, suffix='!') == expected
True
Outer signature and modifiers

The wrapped callable's parameter list and return annotation define the outer interface of the template.

fixed_params permanently supplies selected callable parameters.

param_key_map renames selected callable parameters to external keyword names that callers or parent flows use when supplying inputs.

iterate_over_data_files, output_dataset_param, and output_dataset_cls adapt that outer interface for dataset-wise iteration.

When iterate_over_data_files=True and the inner first parameter is annotated as Model[T], callers see an outer dataset: Dataset[Model[T]] parameter and the outer return type becomes a dataset of the per-item return type. The inner callable still receives one model object at a time.

result_key wraps the returned value in a single-key dictionary, which is especially useful when a downstream DAG step should receive the result under a predictable name.

Examples:

>>> # With modifiers
>>> import omnipy as om
>>> @om.TaskTemplate()
... def plus_other(number: int, other: int) -> int:
...     return number + other

>>> plus_one = plus_other.refine(fixed_params={'other': 1})
>>> plus_one.run(4)
5
>>> plus_x = plus_other.refine(param_key_map={'other': 'x'})
>>> plus_x.run(4, x=3)
7
>>> plus_one_dict = plus_one.refine(result_key='number')
>>> plus_one_dict.run(4)
{'number': 5}

Examples:

>>> # With dataset-wise iteration
>>> import omnipy as om
>>> class TextModel(om.Model[str]): ...
>>> class TextDataset(om.Dataset[TextModel]): ...

>>> @om.TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TextDataset)
... def add_suffix(
...     data_file: TextModel,
...     suffix: str,
... ) -> TextModel:
...     return f'{data_file.content}{suffix}'

>>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
>>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
>>> add_suffix.run(text_files, suffix='!') == expected
True
Tasks and flows

Tasks are terminal jobs: they wrap one callable and execute one compute step.

Flows are orchestration jobs: they may contain child tasks and child flows, so larger pipelines can be assembled hierarchically from smaller reusable pieces.

Lifecycle

Apply a template with apply() to create a runnable job with engine decorators and current config attached. Call the resulting applied job with runtime arguments.

Use run() as a shorthand for apply() followed immediately by calling the applied job.

Use refine() to reuse a template while changing configuration such as name, fixed_params, or param_key_map.

Use revise() on an applied job to reconstruct a template from that job's current configuration.

Examples:

>>> import omnipy as om
>>> @om.TaskTemplate()
... def plus_one(number: int) -> int:
...     return number + 1

>>> plus_one.run(1)
2
>>> applied_job = plus_one.apply()
>>> applied_job(2)
3
>>> refined_template = plus_one.refine(name='plus_one_renamed')
>>> revised_template = applied_job.revise()

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

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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
class TaskTemplateCore(
        FuncArgJobBase[
            IsTaskTemplate[_CallP, _RetT],
            IsTask[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        JobTemplateMixin[
            IsTaskTemplate[_CallP, _RetT],
            IsTask[_CallP, _RetT],
            _CallP,
            _RetT,
        ],
        TaskBase,
        Generic[_CallP, _RetT],
):
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Implement the core template behavior for tasks.
    #
    # {{TASK_TEMPLATE_DESCRIPTION}}
    #
    # Instances are normally produced through the [TaskTemplate][] decorator
    # factory rather than by direct construction.
    #
    """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.

    ### Decorator usage

    Apply the template factory as a decorator to a Python callable.
    The wrapped callable becomes a reusable job template whose public outer
    signature is visible to template users and to the applied jobs created
    from it.

    ### Wrapped callable

    The wrapped callable defines both the implementation and the public
    outer signature of the task.

    Examples:
        >>> import omnipy as om
        >>> class TextModel(om.Model[str]):
        ...     ...

        >>> class TextDataset(om.Dataset[TextModel]):
        ...     ...

        >>> @om.TaskTemplate()
        ... def add_suffix(
        ...     dataset: TextDataset,
        ...     suffix: str,
        ... ) -> TextModel:
        ...     for data_file_name, value in dataset.items():
        ...         dataset[data_file_name] = f'{value}{suffix}'
        ...     return dataset

        >>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
        >>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
        >>> add_suffix.run(text_files, suffix='!') == expected
        True

    ### Outer signature and modifiers

    The wrapped callable's parameter list and return annotation define the
    outer interface of the template.

    ``fixed_params`` permanently supplies selected callable parameters.

    ``param_key_map`` renames selected callable parameters to external
    keyword names that callers or parent flows use when supplying inputs.

    ``iterate_over_data_files``, ``output_dataset_param``, and
    ``output_dataset_cls`` adapt that outer interface for dataset-wise
    iteration.

    When ``iterate_over_data_files=True`` and the inner first parameter is
    annotated as ``Model[T]``, callers see an outer
    ``dataset: Dataset[Model[T]]`` parameter and the outer return type
    becomes a dataset of the per-item return type. The inner callable still
    receives one model object at a time.

    ``result_key`` wraps the returned value in a single-key dictionary,
    which is especially useful when a downstream DAG step should receive
    the result under a predictable name.

    Examples:
        >>> # With modifiers
        >>> import omnipy as om
        >>> @om.TaskTemplate()
        ... def plus_other(number: int, other: int) -> int:
        ...     return number + other

        >>> plus_one = plus_other.refine(fixed_params={'other': 1})
        >>> plus_one.run(4)
        5
        >>> plus_x = plus_other.refine(param_key_map={'other': 'x'})
        >>> plus_x.run(4, x=3)
        7
        >>> plus_one_dict = plus_one.refine(result_key='number')
        >>> plus_one_dict.run(4)
        {'number': 5}

    Examples:
        >>> # With dataset-wise iteration
        >>> import omnipy as om
        >>> class TextModel(om.Model[str]): ...
        >>> class TextDataset(om.Dataset[TextModel]): ...

        >>> @om.TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TextDataset)
        ... def add_suffix(
        ...     data_file: TextModel,
        ...     suffix: str,
        ... ) -> TextModel:
        ...     return f'{data_file.content}{suffix}'

        >>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
        >>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
        >>> add_suffix.run(text_files, suffix='!') == expected
        True

    ### Tasks and flows

    Tasks are terminal jobs: they wrap one callable and execute one compute
    step.

    Flows are orchestration jobs: they may contain child tasks and child
    flows, so larger pipelines can be assembled hierarchically from smaller
    reusable pieces.

    ### Lifecycle

    Apply a template with [`apply()`][omnipy.compute._job.JobTemplateMixin.apply]
    to create a runnable job with engine decorators and current config attached.
    Call the resulting applied job with runtime arguments.

    Use [`run()`][omnipy.compute._job.JobTemplateMixin.run] as a shorthand for
    ``apply()`` followed immediately by calling the applied job.

    Use [`refine()`][omnipy.compute._job.JobTemplateMixin.refine] to reuse a
    template while changing configuration such as ``name``, ``fixed_params``,
    or ``param_key_map``.

    Use [`revise()`][omnipy.compute._job.JobMixin.revise] on an applied job to
    reconstruct a template from that job's current configuration.

    Examples:
        >>> import omnipy as om
        >>> @om.TaskTemplate()
        ... def plus_one(number: int) -> int:
        ...     return number + 1

        >>> plus_one.run(1)
        2
        >>> applied_job = plus_one.apply()
        >>> applied_job(2)
        3
        >>> refined_template = plus_one.refine(name='plus_one_renamed')
        >>> revised_template = applied_job.revise()

    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 
153
154
155
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 
 97
 98
 99
100
101
102
103
104
105
@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

Referenced by:
Source code in src/omnipy/compute/_job.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
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 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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.

Referenced by:
Source code in src/omnipy/compute/_job.py 
298
299
300
301
302
303
304
305
306
307
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 
155
156
157
158
159
160
@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

Referenced by:
Source code in src/omnipy/compute/_job.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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)

TaskTemplate 

TaskTemplate(
    *,
    iterate_over_data_files: Literal[True],
    output_dataset_cls: type[_RetDatasetClsT],
    **kwargs: Unpack[JobCommonKwargs],
) -> TaskTemplateIterWithDatasetClsDecorator[_RetDatasetClsT]

TaskTemplate(
    *,
    iterate_over_data_files: Literal[True],
    output_dataset_cls: None = None,
    **kwargs: Unpack[JobCommonKwargs],
) -> TaskTemplateIterDecorator

TaskTemplate(
    *,
    iterate_over_data_files: Literal[False] = False,
    output_dataset_cls: type[IsDataset] | None = None,
    **kwargs: Unpack[JobCommonKwargs],
) -> TaskTemplatePlainDecorator

TaskTemplate(
    *,
    iterate_over_data_files: bool,
    output_dataset_cls: type[_RetDatasetClsT] | None = None,
    **kwargs: Unpack[JobCommonKwargs],
) -> (
    TaskTemplatePlainDecorator
    | TaskTemplateIterDecorator
    | TaskTemplateIterWithDatasetClsDecorator[_RetDatasetClsT]
)

Decorator-style factory for defining reusable callable-backed 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.

Decorator usage

Apply the template factory as a decorator to a Python callable. The wrapped callable becomes a reusable job template whose public outer signature is visible to template users and to the applied jobs created from it.

Wrapped callable

The wrapped callable defines both the implementation and the public outer signature of the task.

Examples:

>>> import omnipy as om
>>> class TextModel(om.Model[str]):
...     ...

>>> class TextDataset(om.Dataset[TextModel]):
...     ...

>>> @om.TaskTemplate()
... def add_suffix(
...     dataset: TextDataset,
...     suffix: str,
... ) -> TextModel:
...     for data_file_name, value in dataset.items():
...         dataset[data_file_name] = f'{value}{suffix}'
...     return dataset

>>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
>>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
>>> add_suffix.run(text_files, suffix='!') == expected
True
Outer signature and modifiers

The wrapped callable's parameter list and return annotation define the outer interface of the template.

fixed_params permanently supplies selected callable parameters.

param_key_map renames selected callable parameters to external keyword names that callers or parent flows use when supplying inputs.

iterate_over_data_files, output_dataset_param, and output_dataset_cls adapt that outer interface for dataset-wise iteration.

When iterate_over_data_files=True and the inner first parameter is annotated as Model[T], callers see an outer dataset: Dataset[Model[T]] parameter and the outer return type becomes a dataset of the per-item return type. The inner callable still receives one model object at a time.

result_key wraps the returned value in a single-key dictionary, which is especially useful when a downstream DAG step should receive the result under a predictable name.

Examples:

>>> # With modifiers
>>> import omnipy as om
>>> @om.TaskTemplate()
... def plus_other(number: int, other: int) -> int:
...     return number + other

>>> plus_one = plus_other.refine(fixed_params={'other': 1})
>>> plus_one.run(4)
5
>>> plus_x = plus_other.refine(param_key_map={'other': 'x'})
>>> plus_x.run(4, x=3)
7
>>> plus_one_dict = plus_one.refine(result_key='number')
>>> plus_one_dict.run(4)
{'number': 5}

Examples:

>>> # With dataset-wise iteration
>>> import omnipy as om
>>> class TextModel(om.Model[str]): ...
>>> class TextDataset(om.Dataset[TextModel]): ...

>>> @om.TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TextDataset)
... def add_suffix(
...     data_file: TextModel,
...     suffix: str,
... ) -> TextModel:
...     return f'{data_file.content}{suffix}'

>>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
>>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
>>> add_suffix.run(text_files, suffix='!') == expected
True
Tasks and flows

Tasks are terminal jobs: they wrap one callable and execute one compute step.

Flows are orchestration jobs: they may contain child tasks and child flows, so larger pipelines can be assembled hierarchically from smaller reusable pieces.

Lifecycle

Apply a template with apply() to create a runnable job with engine decorators and current config attached. Call the resulting applied job with runtime arguments.

Use run() as a shorthand for apply() followed immediately by calling the applied job.

Use refine() to reuse a template while changing configuration such as name, fixed_params, or param_key_map.

Use revise() on an applied job to reconstruct a template from that job's current configuration.

Examples:

>>> import omnipy as om
>>> @om.TaskTemplate()
... def plus_one(number: int) -> int:
...     return number + 1

>>> plus_one.run(1)
2
>>> applied_job = plus_one.apply()
>>> applied_job(2)
3
>>> refined_template = plus_one.refine(name='plus_one_renamed')
>>> revised_template = applied_job.revise()
PARAMETER DESCRIPTION
name

Name of the job template. If not provided, the name of the wrapped callable is used.

TYPE: str | None DEFAULT: None

iterate_over_data_files

Whether dataset inputs should be processed item-wise.

TYPE: bool DEFAULT: False

output_dataset_param

Optional name of an explicit output-dataset parameter.

TYPE: str | None DEFAULT: None

output_dataset_cls

Optional dataset class to use for iterated outputs.

TYPE: type[IsDataset] | None DEFAULT: None

auto_async

Whether coroutine jobs at the outermost level (not in a flow context) should be automatically run in accordance with context (use existing event loop, if available, otherwise create temporary event loop and run coroutine until completion).

TYPE: bool DEFAULT: True

result_key

Optional key used to wrap the returned result in a dictionary. Especially useful in DAG flows to avoid name collisions.

TYPE: str | None DEFAULT: None

fixed_params

Fixed keyword-argument values for the job. May not target args or *kwargs-style params.

TYPE: Mapping[str, object] | Iterable[tuple[str, object]] | None DEFAULT: None

param_key_map

Mapping from callable parameter names to external keyword names. May not target args or *kwargs-style params.

TYPE: Mapping[str, str] | Iterable[tuple[str, str]] | None DEFAULT: None

persist_outputs

Per-job output-persistence preference.

TYPE: PersistOutputsOptions.Literals DEFAULT: PersistOutputsOptions.FOLLOW_CONFIG

restore_outputs

Per-job output-restore preference.

TYPE: RestoreOutputsOptions.Literals DEFAULT: RestoreOutputsOptions.FOLLOW_CONFIG

**kwargs

Additional constructor keyword overrides.

TYPE: object DEFAULT: {}

Returns: TaskTemplate: New TaskTemplate instance wrapping job_func.

Referenced by:
Source code in src/omnipy/compute/task.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
def TaskTemplate(
    *,
    name: str | None = None,
    iterate_over_data_files: bool = False,
    output_dataset_param: str | None = None,
    output_dataset_cls: type[IsDataset] | None = None,
    auto_async: bool = True,
    result_key: str | None = None,
    fixed_params: Mapping[str, object] | Iterable[tuple[str, object]] | None = None,
    param_key_map: Mapping[str, str] | Iterable[tuple[str, str]] | None = None,
    persist_outputs: PersistOutputsOptions.Literals = PersistOutputsOptions.FOLLOW_CONFIG,
    restore_outputs: RestoreOutputsOptions.Literals = RestoreOutputsOptions.FOLLOW_CONFIG,
    **kwargs: object,
) -> Any:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Decorator-style factory for defining reusable callable-backed tasks.
    #
    # {{TASK_TEMPLATE_DESCRIPTION}}
    #
    # Args:
    #     {{JOB_TEMPLATE_SHARED_KWARG_DOCS}}
    # Returns:
    #     TaskTemplate: New TaskTemplate instance wrapping ``job_func``.
    """Decorator-style factory for defining reusable callable-backed 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.

    ### Decorator usage

    Apply the template factory as a decorator to a Python callable.
    The wrapped callable becomes a reusable job template whose public outer
    signature is visible to template users and to the applied jobs created
    from it.

    ### Wrapped callable

    The wrapped callable defines both the implementation and the public
    outer signature of the task.

    Examples:
        >>> import omnipy as om
        >>> class TextModel(om.Model[str]):
        ...     ...

        >>> class TextDataset(om.Dataset[TextModel]):
        ...     ...

        >>> @om.TaskTemplate()
        ... def add_suffix(
        ...     dataset: TextDataset,
        ...     suffix: str,
        ... ) -> TextModel:
        ...     for data_file_name, value in dataset.items():
        ...         dataset[data_file_name] = f'{value}{suffix}'
        ...     return dataset

        >>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
        >>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
        >>> add_suffix.run(text_files, suffix='!') == expected
        True

    ### Outer signature and modifiers

    The wrapped callable's parameter list and return annotation define the
    outer interface of the template.

    ``fixed_params`` permanently supplies selected callable parameters.

    ``param_key_map`` renames selected callable parameters to external
    keyword names that callers or parent flows use when supplying inputs.

    ``iterate_over_data_files``, ``output_dataset_param``, and
    ``output_dataset_cls`` adapt that outer interface for dataset-wise
    iteration.

    When ``iterate_over_data_files=True`` and the inner first parameter is
    annotated as ``Model[T]``, callers see an outer
    ``dataset: Dataset[Model[T]]`` parameter and the outer return type
    becomes a dataset of the per-item return type. The inner callable still
    receives one model object at a time.

    ``result_key`` wraps the returned value in a single-key dictionary,
    which is especially useful when a downstream DAG step should receive
    the result under a predictable name.

    Examples:
        >>> # With modifiers
        >>> import omnipy as om
        >>> @om.TaskTemplate()
        ... def plus_other(number: int, other: int) -> int:
        ...     return number + other

        >>> plus_one = plus_other.refine(fixed_params={'other': 1})
        >>> plus_one.run(4)
        5
        >>> plus_x = plus_other.refine(param_key_map={'other': 'x'})
        >>> plus_x.run(4, x=3)
        7
        >>> plus_one_dict = plus_one.refine(result_key='number')
        >>> plus_one_dict.run(4)
        {'number': 5}

    Examples:
        >>> # With dataset-wise iteration
        >>> import omnipy as om
        >>> class TextModel(om.Model[str]): ...
        >>> class TextDataset(om.Dataset[TextModel]): ...

        >>> @om.TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TextDataset)
        ... def add_suffix(
        ...     data_file: TextModel,
        ...     suffix: str,
        ... ) -> TextModel:
        ...     return f'{data_file.content}{suffix}'

        >>> text_files = TextDataset({'a': 'hi', 'b': 'bye'})
        >>> expected = TextDataset({'a': 'hi!', 'b': 'bye!'})
        >>> add_suffix.run(text_files, suffix='!') == expected
        True

    ### Tasks and flows

    Tasks are terminal jobs: they wrap one callable and execute one compute
    step.

    Flows are orchestration jobs: they may contain child tasks and child
    flows, so larger pipelines can be assembled hierarchically from smaller
    reusable pieces.

    ### Lifecycle

    Apply a template with [`apply()`][omnipy.compute._job.JobTemplateMixin.apply]
    to create a runnable job with engine decorators and current config attached.
    Call the resulting applied job with runtime arguments.

    Use [`run()`][omnipy.compute._job.JobTemplateMixin.run] as a shorthand for
    ``apply()`` followed immediately by calling the applied job.

    Use [`refine()`][omnipy.compute._job.JobTemplateMixin.refine] to reuse a
    template while changing configuration such as ``name``, ``fixed_params``,
    or ``param_key_map``.

    Use [`revise()`][omnipy.compute._job.JobMixin.revise] on an applied job to
    reconstruct a template from that job's current configuration.

    Examples:
        >>> import omnipy as om
        >>> @om.TaskTemplate()
        ... def plus_one(number: int) -> int:
        ...     return number + 1

        >>> plus_one.run(1)
        2
        >>> applied_job = plus_one.apply()
        >>> applied_job(2)
        3
        >>> refined_template = plus_one.refine(name='plus_one_renamed')
        >>> revised_template = applied_job.revise()

    Args:
        name: Name of the job template. If not provided, the name of the
            wrapped callable is used.
        iterate_over_data_files: Whether dataset inputs should be
            processed item-wise.
        output_dataset_param: Optional name of an explicit
            output-dataset parameter.
        output_dataset_cls: Optional dataset class to use for iterated
            outputs.
        auto_async: Whether coroutine jobs at the outermost level (not
            in a flow context) should be automatically run in accordance
            with context (use existing event loop, if available,
            otherwise create temporary event loop and run coroutine
            until completion).
        result_key: Optional key used to wrap the returned result in a
            dictionary. Especially useful in DAG flows to avoid name
            collisions.
        fixed_params: Fixed keyword-argument values for the job. May not
            target *args or **kwargs-style params.
        param_key_map: Mapping from callable parameter names to external
            keyword names. May not target *args or **kwargs-style
            params.
        persist_outputs: Per-job output-persistence preference.
        restore_outputs: Per-job output-restore preference.
        **kwargs: Additional constructor keyword overrides.
    Returns:
        TaskTemplate: New TaskTemplate instance wrapping ``job_func``."""
    ret = _TaskTemplateFactory(
        name=name,
        iterate_over_data_files=iterate_over_data_files,
        output_dataset_param=output_dataset_param,
        output_dataset_cls=output_dataset_cls,
        auto_async=auto_async,
        result_key=result_key,
        fixed_params=fixed_params,
        param_key_map=param_key_map,
        persist_outputs=persist_outputs,
        restore_outputs=restore_outputs,
        **kwargs,
    )
    return ret