omnipy.engine.run_spec

Engine-facing run-spec adapters for tasks and flows.

Run-spec objects expose a uniform view of Omnipy jobs so execution engines can initialize them, wrap their callables, and execute them without depending on the full job implementation details.

CLASS	DESCRIPTION
ChildJobListArgFlowRunSpec	Base run spec for flows defined by ordered child-job templates.
DagFlowRunSpec	Run spec for DAG flows that route named results into downstream inputs.
FlowRunSpec	Base run-spec adapter for flow jobs.
FuncFlowRunSpec	Run spec for callable-backed flows that execute a single wrapped callable.
JobRunSpec	Base adapter exposing engine-facing metadata and callables for a job.
LinearFlowRunSpec	Run spec for linear flows that pipe each child result into the next.
TaskRunSpec	Run-spec adapter for applied tasks.

ChildJobListArgFlowRunSpec

Bases: FlowRunSpec, ABC

              flowchart BT
              omnipy.engine.run_spec.ChildJobListArgFlowRunSpec[ChildJobListArgFlowRunSpec]
              omnipy.engine.run_spec.FlowRunSpec[FlowRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.FlowRunSpec --> omnipy.engine.run_spec.ChildJobListArgFlowRunSpec
                                omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.FlowRunSpec
                



              click omnipy.engine.run_spec.ChildJobListArgFlowRunSpec href "" "omnipy.engine.run_spec.ChildJobListArgFlowRunSpec"
              click omnipy.engine.run_spec.FlowRunSpec href "" "omnipy.engine.run_spec.FlowRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Base run spec for flows defined by ordered child-job templates.

Subclassed by:

• Reference Code reference omnipy engine  run_spec
  •  DagFlowRunSpec
  •  LinearFlowRunSpec

METHOD	DESCRIPTION
__init__
create_default_run_callable	Wrap the flow callable with its declared callable shape and flow context.
get_bound_args	Proxies argument binding to the wrapped flow callable.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
child_job_templates	Proxies the wrapped flow's ordered child templates. TYPE: tuple[ChildJobTemplateLike, ...]
flow_context	Proxies the wrapped flow's nested execution context.
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class ChildJobListArgFlowRunSpec(FlowRunSpec, ABC):
    """Base run spec for flows defined by ordered child-job templates."""
    @property
    def child_job_templates(self) -> tuple[ChildJobTemplateLike, ...]:
        """Proxies the wrapped flow's ordered child templates.

        See [`IsChildJobListArgJobBase.child_job_templates`]
        [omnipy.shared.protocols.compute.job.IsChildJobListArgJobBase.child_job_templates].
        """
        flow = cast(IsChildJobListArgJob, self._job)
        return flow.child_job_templates

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

child_job_templates property

child_job_templates: tuple[ChildJobTemplateLike, ...]

Proxies the wrapped flow's ordered child templates.

See IsChildJobListArgJobBase.child_job_templates.

flow_context property

flow_context

Proxies the wrapped flow's nested execution context.

See IsFlow.flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Wrap the flow callable with its declared callable shape and flow context.

RETURNS	DESCRIPTION
Callable	Engine-facing flow callable that enters flow_context. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Wrap the flow callable with its declared callable shape and flow context.

    Returns:
        Callable: Engine-facing flow callable that enters ``flow_context``.
    """
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        self.param_signatures,
        self.return_type,
        self.callable_type,
        context_factory=lambda: self.flow_context,
    )

get_bound_args

get_bound_args(*args: object, **kwargs: object) -> BoundArguments

Proxies argument binding to the wrapped flow callable.

See IsFuncArgJobBase.get_bound_args.

Source code in src/omnipy/engine/run_spec.py

def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    """Proxies argument binding to the wrapped flow callable.

    See [`IsFuncArgJobBase.get_bound_args`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
    """
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

DagFlowRunSpec

Bases: ChildJobListArgFlowRunSpec

              flowchart BT
              omnipy.engine.run_spec.DagFlowRunSpec[DagFlowRunSpec]
              omnipy.engine.run_spec.ChildJobListArgFlowRunSpec[ChildJobListArgFlowRunSpec]
              omnipy.engine.run_spec.FlowRunSpec[FlowRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.ChildJobListArgFlowRunSpec --> omnipy.engine.run_spec.DagFlowRunSpec
                                omnipy.engine.run_spec.FlowRunSpec --> omnipy.engine.run_spec.ChildJobListArgFlowRunSpec
                                omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.FlowRunSpec
                




              click omnipy.engine.run_spec.DagFlowRunSpec href "" "omnipy.engine.run_spec.DagFlowRunSpec"
              click omnipy.engine.run_spec.ChildJobListArgFlowRunSpec href "" "omnipy.engine.run_spec.ChildJobListArgFlowRunSpec"
              click omnipy.engine.run_spec.FlowRunSpec href "" "omnipy.engine.run_spec.FlowRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Run spec for DAG flows that route named results into downstream inputs.

Used by:

• Reference Code reference omnipy engine  job_runner  JOB_TYPE_TO_RUN_DEF

METHOD	DESCRIPTION
__init__
create_default_run_callable	Wrap the flow callable with its declared callable shape and flow context.
get_bound_args	Proxies argument binding to the wrapped flow callable.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
child_job_templates	Proxies the wrapped flow's ordered child templates. TYPE: tuple[ChildJobTemplateLike, ...]
flow_context	Proxies the wrapped flow's nested execution context.
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class DagFlowRunSpec(ChildJobListArgFlowRunSpec):
    """Run spec for DAG flows that route named results into downstream inputs."""
    def _create_default_run_callable(self) -> Callable:  # noqa: C901
        def _run_all_dag_tasks(
            *args: object,
            **kwargs: object,
        ) -> Generator[object, object, object]:
            results: dict[str, object] = {}
            result = None

            for i, job in enumerate(self.child_job_templates):
                assert not inspect.isclass(job)

                if i == 0:
                    results = self.get_bound_args(*args, **kwargs).arguments

                params = _collect_matching_kwargs_for_job(job, results)
                result = job(**params)

                result = yield result

                if isinstance(result, dict) and len(result) > 0:
                    results.update(result)
                else:
                    results[job.name] = result

            return result

        if _has_any_async_coroutine_jobs(self):

            async def _async_inner_run_dag_flow(*args: object, **kwargs: object):
                async def _resolve_awaitable_dag_task_result(result: object) -> object:
                    result = await resolve(result)

                    if isinstance(result, dict) and len(result) > 0:
                        result = {key: await resolve(val) for key, val in result.items()}

                    return result

                return await _drain_async_results(
                    _run_all_dag_tasks(*args, **kwargs),
                    _resolve_awaitable_dag_task_result,
                )

            return _async_inner_run_dag_flow
        else:

            def _sync_inner_run_dag_flow(*args: object, **kwargs: object):
                return _drain_sync_results(_run_all_dag_tasks(*args, **kwargs))

            return _sync_inner_run_dag_flow

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

child_job_templates property

child_job_templates: tuple[ChildJobTemplateLike, ...]

Proxies the wrapped flow's ordered child templates.

See IsChildJobListArgJobBase.child_job_templates.

flow_context property

flow_context

Proxies the wrapped flow's nested execution context.

See IsFlow.flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Wrap the flow callable with its declared callable shape and flow context.

RETURNS	DESCRIPTION
Callable	Engine-facing flow callable that enters flow_context. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Wrap the flow callable with its declared callable shape and flow context.

    Returns:
        Callable: Engine-facing flow callable that enters ``flow_context``.
    """
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        self.param_signatures,
        self.return_type,
        self.callable_type,
        context_factory=lambda: self.flow_context,
    )

get_bound_args

get_bound_args(*args: object, **kwargs: object) -> BoundArguments

Proxies argument binding to the wrapped flow callable.

See IsFuncArgJobBase.get_bound_args.

Source code in src/omnipy/engine/run_spec.py

def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    """Proxies argument binding to the wrapped flow callable.

    See [`IsFuncArgJobBase.get_bound_args`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
    """
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

FlowRunSpec

Bases: JobRunSpec, ABC

              flowchart BT
              omnipy.engine.run_spec.FlowRunSpec[FlowRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.FlowRunSpec
                


              click omnipy.engine.run_spec.FlowRunSpec href "" "omnipy.engine.run_spec.FlowRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Base run-spec adapter for flow jobs.

Subclassed by:

• Reference Code reference omnipy engine  run_spec
  •  ChildJobListArgFlowRunSpec
  •  FuncFlowRunSpec

METHOD	DESCRIPTION
__init__
create_default_run_callable	Wrap the flow callable with its declared callable shape and flow context.
get_bound_args	Proxies argument binding to the wrapped flow callable.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
flow_context	Proxies the wrapped flow's nested execution context.
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class FlowRunSpec(JobRunSpec, ABC):
    """Base run-spec adapter for flow jobs."""
    @property
    def flow_context(self):
        """Proxies the wrapped flow's nested execution context.

        See [`IsFlow.flow_context`][omnipy.shared.protocols.compute.job.IsFlow.flow_context].
        """
        flow = cast(IsAnyFlow, self._job)
        return flow.flow_context

    def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
        """Proxies argument binding to the wrapped flow callable.

        See [`IsFuncArgJobBase.get_bound_args`]
        [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
        """
        flow = cast(IsAnyFlow, self._job)
        return flow.get_bound_args(*args, **kwargs)

    def create_default_run_callable(self) -> Callable:
        """Wrap the flow callable with its declared callable shape and flow context.

        Returns:
            Callable: Engine-facing flow callable that enters ``flow_context``.
        """
        return decorate_callable_by_type(
            self._create_default_run_callable(),
            self.param_signatures,
            self.return_type,
            self.callable_type,
            context_factory=lambda: self.flow_context,
        )

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

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

flow_context property

flow_context

Proxies the wrapped flow's nested execution context.

See IsFlow.flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Wrap the flow callable with its declared callable shape and flow context.

RETURNS	DESCRIPTION
Callable	Engine-facing flow callable that enters flow_context. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Wrap the flow callable with its declared callable shape and flow context.

    Returns:
        Callable: Engine-facing flow callable that enters ``flow_context``.
    """
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        self.param_signatures,
        self.return_type,
        self.callable_type,
        context_factory=lambda: self.flow_context,
    )

get_bound_args

get_bound_args(*args: object, **kwargs: object) -> BoundArguments

Proxies argument binding to the wrapped flow callable.

See IsFuncArgJobBase.get_bound_args.

Source code in src/omnipy/engine/run_spec.py

def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    """Proxies argument binding to the wrapped flow callable.

    See [`IsFuncArgJobBase.get_bound_args`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
    """
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

FuncFlowRunSpec

Bases: FlowRunSpec

              flowchart BT
              omnipy.engine.run_spec.FuncFlowRunSpec[FuncFlowRunSpec]
              omnipy.engine.run_spec.FlowRunSpec[FlowRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.FlowRunSpec --> omnipy.engine.run_spec.FuncFlowRunSpec
                                omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.FlowRunSpec
                



              click omnipy.engine.run_spec.FuncFlowRunSpec href "" "omnipy.engine.run_spec.FuncFlowRunSpec"
              click omnipy.engine.run_spec.FlowRunSpec href "" "omnipy.engine.run_spec.FlowRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Run spec for callable-backed flows that execute a single wrapped callable.

Used by:

• Reference Code reference omnipy engine  job_runner  JOB_TYPE_TO_RUN_DEF

METHOD	DESCRIPTION
__init__
create_default_run_callable	Wrap the flow callable with its declared callable shape and flow context.
get_bound_args	Proxies argument binding to the wrapped flow callable.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
flow_context	Proxies the wrapped flow's nested execution context.
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class FuncFlowRunSpec(FlowRunSpec):
    """Run spec for callable-backed flows that execute a single wrapped callable."""
    def _create_default_run_callable(self) -> Callable:
        return self._run_callable

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

flow_context property

flow_context

Proxies the wrapped flow's nested execution context.

See IsFlow.flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Wrap the flow callable with its declared callable shape and flow context.

RETURNS	DESCRIPTION
Callable	Engine-facing flow callable that enters flow_context. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Wrap the flow callable with its declared callable shape and flow context.

    Returns:
        Callable: Engine-facing flow callable that enters ``flow_context``.
    """
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        self.param_signatures,
        self.return_type,
        self.callable_type,
        context_factory=lambda: self.flow_context,
    )

get_bound_args

get_bound_args(*args: object, **kwargs: object) -> BoundArguments

Proxies argument binding to the wrapped flow callable.

See IsFuncArgJobBase.get_bound_args.

Source code in src/omnipy/engine/run_spec.py

def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    """Proxies argument binding to the wrapped flow callable.

    See [`IsFuncArgJobBase.get_bound_args`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
    """
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

JobRunSpec

Bases: ABC

              flowchart BT
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

              

              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Base adapter exposing engine-facing metadata and callables for a job.

Returned by:

• Reference Code reference omnipy engine  job_runner  JobRunDef  run_spec

Subclassed by:

• Reference Code reference omnipy engine  run_spec
  •  FlowRunSpec
  •  TaskRunSpec

METHOD	DESCRIPTION
__init__
create_default_run_callable	Build the default callable that engines should initialize and execute.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class JobRunSpec(ABC):
    """Base adapter exposing engine-facing metadata and callables for a job."""
    def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
        self._job = job
        self._run_callable = run_callable

    @property
    def name(self) -> str:
        """Proxies to the wrapped job's display name.

        See [`IsUniquelyNamedJob.name`]
        [omnipy.shared.protocols.compute.mixins.IsUniquelyNamedJob.name].
        """
        return self._job.name

    @property
    def unique_name(self) -> str:
        """Proxies to the wrapped job's unique registry name.

        See [`IsUniquelyNamedJob.unique_name`]
        [omnipy.shared.protocols.compute.mixins.IsUniquelyNamedJob.unique_name].
        """
        return self._job.unique_name

    @property
    def unique_run_slug(self) -> str:
        """Proxies to the wrapped job's run-specific slug.

        See [`IsUniquelyNamedJob.unique_run_slug`]
        [omnipy.shared.protocols.compute.mixins.IsUniquelyNamedJob.unique_run_slug].
        """
        return self._job.unique_run_slug

    @property
    def param_signatures(self) -> MappingProxyType[str, inspect.Parameter]:
        """Proxies to the wrapped job's callable parameter metadata.

        See [`IsFuncArgJobBase.param_signatures`]
        [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.param_signatures].
        """
        return self._job.param_signatures

    @property
    def return_type(self) -> type:
        """Proxies to the wrapped job's annotated return type.

        See [`IsFuncArgJobBase.return_type`]
        [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.return_type].
        """
        return self._job.return_type

    @property
    def callable_type(self) -> CallableType.Literals:
        """Proxies to the wrapped job's callable-shape classification.

        See [`IsFuncArgJobBase.callable_type`]
        [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.callable_type].
        """
        return self._job.callable_type

    def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
        """Proxies log messages to the wrapped job.

        See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
        """
        self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

    @abstractmethod
    def create_default_run_callable(self) -> Callable:
        """Build the default callable that engines should initialize and execute.

        Returns:
            Callable: Engine-facing callable for running the wrapped job.
        """
        ...

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable abstractmethod

create_default_run_callable() -> Callable

Build the default callable that engines should initialize and execute.

RETURNS	DESCRIPTION
Callable	Engine-facing callable for running the wrapped job. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

@abstractmethod
def create_default_run_callable(self) -> Callable:
    """Build the default callable that engines should initialize and execute.

    Returns:
        Callable: Engine-facing callable for running the wrapped job.
    """
    ...

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

LinearFlowRunSpec

Bases: ChildJobListArgFlowRunSpec

              flowchart BT
              omnipy.engine.run_spec.LinearFlowRunSpec[LinearFlowRunSpec]
              omnipy.engine.run_spec.ChildJobListArgFlowRunSpec[ChildJobListArgFlowRunSpec]
              omnipy.engine.run_spec.FlowRunSpec[FlowRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.ChildJobListArgFlowRunSpec --> omnipy.engine.run_spec.LinearFlowRunSpec
                                omnipy.engine.run_spec.FlowRunSpec --> omnipy.engine.run_spec.ChildJobListArgFlowRunSpec
                                omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.FlowRunSpec
                




              click omnipy.engine.run_spec.LinearFlowRunSpec href "" "omnipy.engine.run_spec.LinearFlowRunSpec"
              click omnipy.engine.run_spec.ChildJobListArgFlowRunSpec href "" "omnipy.engine.run_spec.ChildJobListArgFlowRunSpec"
              click omnipy.engine.run_spec.FlowRunSpec href "" "omnipy.engine.run_spec.FlowRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Run spec for linear flows that pipe each child result into the next.

Used by:

• Reference Code reference omnipy engine  job_runner  JOB_TYPE_TO_RUN_DEF

METHOD	DESCRIPTION
__init__
create_default_run_callable	Wrap the flow callable with its declared callable shape and flow context.
get_bound_args	Proxies argument binding to the wrapped flow callable.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
child_job_templates	Proxies the wrapped flow's ordered child templates. TYPE: tuple[ChildJobTemplateLike, ...]
flow_context	Proxies the wrapped flow's nested execution context.
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class LinearFlowRunSpec(ChildJobListArgFlowRunSpec):
    """Run spec for linear flows that pipe each child result into the next."""
    def _create_default_run_callable(self) -> Callable:
        def _run_all_linear_tasks(
            *args: object,
            **kwargs: object,
        ) -> Generator[object, object, object]:
            flow_kwargs = self.get_bound_args(*args, **kwargs).arguments
            result = None
            for i, job in enumerate(self.child_job_templates):
                assert not inspect.isclass(job)
                call_args = args if i == 0 else (result,)
                task_kwargs = _collect_matching_kwargs_for_job(job, flow_kwargs, len(call_args))
                result = job(*call_args, **task_kwargs)

                result = yield result

                args = (result,)

            return result

        if _has_any_async_coroutine_jobs(self):

            async def _async_inner_run_linear_flow(*args: object, **kwargs: object):
                return await _drain_async_results(_run_all_linear_tasks(*args, **kwargs))

            return _async_inner_run_linear_flow
        else:

            def _sync_inner_run_linear_flow(*args: object, **kwargs: object):
                return _drain_sync_results(_run_all_linear_tasks(*args, **kwargs))

            return _sync_inner_run_linear_flow

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

child_job_templates property

child_job_templates: tuple[ChildJobTemplateLike, ...]

Proxies the wrapped flow's ordered child templates.

See IsChildJobListArgJobBase.child_job_templates.

flow_context property

flow_context

Proxies the wrapped flow's nested execution context.

See IsFlow.flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Wrap the flow callable with its declared callable shape and flow context.

RETURNS	DESCRIPTION
Callable	Engine-facing flow callable that enters flow_context. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Wrap the flow callable with its declared callable shape and flow context.

    Returns:
        Callable: Engine-facing flow callable that enters ``flow_context``.
    """
    return decorate_callable_by_type(
        self._create_default_run_callable(),
        self.param_signatures,
        self.return_type,
        self.callable_type,
        context_factory=lambda: self.flow_context,
    )

get_bound_args

get_bound_args(*args: object, **kwargs: object) -> BoundArguments

Proxies argument binding to the wrapped flow callable.

See IsFuncArgJobBase.get_bound_args.

Source code in src/omnipy/engine/run_spec.py

def get_bound_args(self, *args: object, **kwargs: object) -> BoundArguments:
    """Proxies argument binding to the wrapped flow callable.

    See [`IsFuncArgJobBase.get_bound_args`]
    [omnipy.shared.protocols.compute.job.IsFuncArgJobBase.get_bound_args].
    """
    flow = cast(IsAnyFlow, self._job)
    return flow.get_bound_args(*args, **kwargs)

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)

TaskRunSpec

Bases: JobRunSpec

              flowchart BT
              omnipy.engine.run_spec.TaskRunSpec[TaskRunSpec]
              omnipy.engine.run_spec.JobRunSpec[JobRunSpec]

                              omnipy.engine.run_spec.JobRunSpec --> omnipy.engine.run_spec.TaskRunSpec
                


              click omnipy.engine.run_spec.TaskRunSpec href "" "omnipy.engine.run_spec.TaskRunSpec"
              click omnipy.engine.run_spec.JobRunSpec href "" "omnipy.engine.run_spec.JobRunSpec"
            

Run-spec adapter for applied tasks.

Used by:

• Reference Code reference omnipy engine  job_runner  JOB_TYPE_TO_RUN_DEF

METHOD	DESCRIPTION
__init__
create_default_run_callable	Return the task callable exactly as supplied to the run spec.
log	Proxies log messages to the wrapped job.

ATTRIBUTE	DESCRIPTION
callable_type	Proxies to the wrapped job's callable-shape classification. TYPE: CallableType.Literals
in_flow_context	Proxies whether the wrapped task is running inside a flow. TYPE: bool
name	Proxies to the wrapped job's display name. TYPE: str
param_signatures	Proxies to the wrapped job's callable parameter metadata. TYPE: MappingProxyType[str, inspect.Parameter]
return_type	Proxies to the wrapped job's annotated return type. TYPE: type
unique_name	Proxies to the wrapped job's unique registry name. TYPE: str
unique_run_slug	Proxies to the wrapped job's run-specific slug. TYPE: str

Source code in src/omnipy/engine/run_spec.py

class TaskRunSpec(JobRunSpec):
    """Run-spec adapter for applied tasks."""
    @property
    def in_flow_context(self) -> bool:
        """Proxies whether the wrapped task is running inside a flow.

        See [`IsJobBase.in_flow_context`]
        [omnipy.shared.protocols.compute.job.IsJobBase.in_flow_context].
        """
        task = cast(IsTask, self._job)
        return task.in_flow_context

    def create_default_run_callable(self) -> Callable:
        """Return the task callable exactly as supplied to the run spec.

        Returns:
            Callable: Underlying task callable to hand to the engine.
        """
        return self._run_callable

callable_type property

callable_type: CallableType.Literals

Proxies to the wrapped job's callable-shape classification.

See IsFuncArgJobBase.callable_type.

in_flow_context property

in_flow_context: bool

Proxies whether the wrapped task is running inside a flow.

See IsJobBase.in_flow_context.

name property

name: str

Proxies to the wrapped job's display name.

See IsUniquelyNamedJob.name.

param_signatures property

param_signatures: MappingProxyType[str, inspect.Parameter]

Proxies to the wrapped job's callable parameter metadata.

See IsFuncArgJobBase.param_signatures.

return_type property

return_type: type

Proxies to the wrapped job's annotated return type.

See IsFuncArgJobBase.return_type.

unique_name property

unique_name: str

Proxies to the wrapped job's unique registry name.

See IsUniquelyNamedJob.unique_name.

unique_run_slug property

unique_run_slug: str

Proxies to the wrapped job's run-specific slug.

See IsUniquelyNamedJob.unique_run_slug.

__init__

__init__(job: IsFuncArgJob, run_callable: Callable) -> None

Source code in src/omnipy/engine/run_spec.py

def __init__(self, job: IsFuncArgJob, run_callable: Callable) -> None:
    self._job = job
    self._run_callable = run_callable

create_default_run_callable

create_default_run_callable() -> Callable

Return the task callable exactly as supplied to the run spec.

RETURNS	DESCRIPTION
Callable	Underlying task callable to hand to the engine. TYPE: Callable

Source code in src/omnipy/engine/run_spec.py

def create_default_run_callable(self) -> Callable:
    """Return the task callable exactly as supplied to the run spec.

    Returns:
        Callable: Underlying task callable to hand to the engine.
    """
    return self._run_callable

log

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

Proxies log messages to the wrapped job.

See CanLog.log.

Source code in src/omnipy/engine/run_spec.py

def log(self, log_msg: str, level: int = INFO, datetime_obj: datetime | None = None) -> None:
    """Proxies log messages to the wrapped job.

    See [`CanLog.log`][omnipy.shared.protocols.hub.log.CanLog.log].
    """
    self._job.log(log_msg, level=level, datetime_obj=datetime_obj)