doot.control.tracker._interface

Type Aliases

Abstract

Enums

`EdgeType_e`	Enum describing the possible edges of the task tracker's task network
`ExecutionPolicy_e`	How the task execution will be ordered

Protocols

`Registry_p`	Base class for protocol classes.
`Network_p`	Base class for protocol classes.
`Queue_p`	Base class for protocol classes.
`WorkflowTracker_p`	Track tasks that have run, need to run, are running,
`WorkflowTracker_i`	Track tasks that have run, need to run, are running,

Classes

`SpecMeta_d`	Registry data for a spec.
`ArtifactMeta_d`
`Registry_d`

Module Contents

doot.control.tracker._interface.Abstract: TypeAlias = T

class doot.control.tracker._interface.EdgeType_e(*args, **kwds)[source]

Bases: enum.Enum

Enum describing the possible edges of the task tracker’s task network

TASK

ARTIFACT_UP

ARTIFACT_DOWN

TASK_CROSS

ARTIFACT_CROSS

default

classmethod artifact_edge_set() → set[EdgeType_e.artifact_edge_set.T][source]

Return type:: set[EdgeType_e.artifact_edge_set.T]

class doot.control.tracker._interface.ExecutionPolicy_e(*args, **kwds)[source]

Bases: enum.Enum

How the task execution will be ordered PRIORITY : Priority Queue with retry, job expansion, dynamic walk of network. DEPTH : No (priority,retry,jobs). basic DFS of the pre-run dependency network BREADTH : No (priority,retry,jobs). basic BFS of the pre-run dependency-network

PRIORITY

DEPTH

BREADTH

default

class doot.control.tracker._interface.Registry_p[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

register_spec( *specs: doot.workflow._interface.TaskSpec_i, ) → None[source]

Parameters:: specs (doot.workflow._interface.TaskSpec_i)
Return type:: None

instantiate_spec( name: Abstract[doot.workflow._interface.TaskName_p], *, force: jgdv.Maybe[bool | int] = None, extra: jgdv.Maybe[dict | jgdv.structs.chainguard.ChainGuard | bool] = None, ) → jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p]][source]

Parameters:

name (Abstract[doot.workflow._interface.TaskName_p])
force (jgdv.Maybe[bool | int])
extra (jgdv.Maybe[dict | jgdv.structs.chainguard.ChainGuard | bool])

Return type:

jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p]]

instantiate_relation( rel: doot.workflow._interface.RelationSpec_i, *, control: Concrete[doot.workflow._interface.TaskName_p], ) → Concrete[doot.workflow._interface.TaskName_p][source]

Parameters:

rel (doot.workflow._interface.RelationSpec_i)
control (Concrete[doot.workflow._interface.TaskName_p])

Return type:

Concrete[doot.workflow._interface.TaskName_p]

make_task( name: Concrete[doot.workflow._interface.TaskName_p], *, task_obj: jgdv.Maybe[doot.workflow._interface.Task_i] = None, parent: jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p]] = None, ) → Concrete[doot.workflow._interface.TaskName_p][source]

Parameters:

name (Concrete[doot.workflow._interface.TaskName_p])
task_obj (jgdv.Maybe[doot.workflow._interface.Task_i])
parent (jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p]])

Return type:

Concrete[doot.workflow._interface.TaskName_p]

verify(*, strict: bool = True) → bool[source]

Parameters:: strict (bool)
Return type:: bool

class doot.control.tracker._interface.Network_p[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

_graph: Any

_root_node: doot.workflow._interface.TaskName_p

succ: collections.abc.Mapping

pred: collections.abc.Mapping

non_expanded: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]

build_network( *, sources: jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]] = None, ) → None[source]

Parameters:: sources (jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]])
Return type:: None

connect(

left: Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i,

right: jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i] = None,

**kwargs: Any,

) → None[source]

Parameters:

left (Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i)
right (jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i])
kwargs (Any)

Return type:

None

validate_network(*, strict: bool = True) → bool[source]

Parameters:: strict (bool)
Return type:: bool

class doot.control.tracker._interface.Queue_p[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

active_set: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]

queue_entry( target: str | doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i, *, from_user: int | bool = False, ) → jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]][source]

Parameters:

target (str | doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i)
from_user (int | bool)

Return type:

jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]]

deque_entry( *, peek: bool = False, ) → Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i[source]

Parameters:: peek (bool)
Return type:: Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i

clear_queue() → None[source]

Return type:: None

class doot.control.tracker._interface.WorkflowTracker_p[source]

Bases: Protocol

Track tasks that have run, need to run, are running, and have failed. Does not execute anything itself

property active: set[doot.workflow._interface.TaskName_p][source]

Return type:: set[doot.workflow._interface.TaskName_p]

property specs: dict[doot.workflow._interface.TaskName_p, SpecMeta_d][source]

Return type:: dict[doot.workflow._interface.TaskName_p, SpecMeta_d]

property artifacts: dict[doot.workflow._interface.Artifact_i, ArtifactMeta_d][source]

Return type:: dict[doot.workflow._interface.Artifact_i, ArtifactMeta_d]

property concrete: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i][source]

Return type:: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]

property abstract: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i][source]

Return type:: set[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]

property network: collections.abc.Mapping[source]

Return type:: collections.abc.Mapping

property is_valid: bool[source]

Return type:: bool

register( *specs: doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec, ) → None[source]

Parameters:: specs (doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec)
Return type:: None

queue( name: str | jgdv.Ident | Concrete[doot.workflow._interface.TaskSpec_i] | doot.workflow._interface.DelayedSpec, *, from_user: int | bool = False, status: jgdv.Maybe[doot.workflow._interface.TaskStatus_e] = None, ) → jgdv.Maybe[Concrete[jgdv.Ident]][source]

Parameters:

name (str | jgdv.Ident | Concrete[doot.workflow._interface.TaskSpec_i] | doot.workflow._interface.DelayedSpec)
from_user (int | bool)
status (jgdv.Maybe[doot.workflow._interface.TaskStatus_e])

Return type:

jgdv.Maybe[Concrete[jgdv.Ident]]

build( *, sources: jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]] = None, ) → None[source]

Parameters:: sources (jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]])
Return type:: None

plan( *, policy: jgdv.Maybe[ExecutionPolicy_e] = None, ) → list[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i][source]

Parameters:: policy (jgdv.Maybe[ExecutionPolicy_e])
Return type:: list[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]

next_for( target: jgdv.Maybe[str | Concrete[jgdv.Ident]] = None, ) → jgdv.Maybe[doot.workflow._interface.Task_p | doot.workflow._interface.Artifact_i][source]

Parameters:: target (jgdv.Maybe[str | Concrete[jgdv.Ident]])
Return type:: jgdv.Maybe[doot.workflow._interface.Task_p | doot.workflow._interface.Artifact_i]

clear() → None[source]

Return type:: None

_instantiate( name: Abstract[doot.workflow._interface.TaskName_p], *, extra: jgdv.Maybe[dict | jgdv.structs.chainguard.ChainGuard | bool] = None, ) → jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p]][source]
_instantiate( rel: doot.workflow._interface.RelationSpec_i, *, control: Concrete[doot.workflow._interface.TaskName_p], ) → Concrete[doot.workflow._interface.TaskName_p]
_instantiate(

name: Concrete[doot.workflow._interface.TaskName_p],

**kwargs: Any,

) → Concrete[doot.workflow._interface.TaskName_p]

_connect(

left: Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i,

right: jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i] = None,

**kwargs: Any,

) → None[source]

Parameters:

left (Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i)
right (jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i])
kwargs (Any)

Return type:

None

_dependency_states_of( focus: doot.workflow._interface.TaskName_p, ) → list[tuple][source]

Parameters:: focus (doot.workflow._interface.TaskName_p)
Return type:: list[tuple]

_successor_states_of( focus: doot.workflow._interface.TaskName_p, ) → list[tuple][source]

Parameters:: focus (doot.workflow._interface.TaskName_p)
Return type:: list[tuple]

class doot.control.tracker._interface.WorkflowTracker_i[source]

Bases: WorkflowTracker_p, Protocol

Track tasks that have run, need to run, are running, and have failed. Does not execute anything itself

_root_node: doot.workflow._interface.TaskName_p

_factory: doot.workflow._interface.TaskFactory_p

_subfactory: doot.workflow._interface.SubTaskFactory_p

_declare_priority: int

_min_priority: int

class doot.control.tracker._interface.SpecMeta_d( *, spec: doot.workflow._interface.TaskSpec_i, )[source]

Registry data for a spec. When spec is abstract, related are the concrete instantiations when spec is concrete, related are the implicit subtasks

blocked_by are the dependencies not mentioned in the spec injection_source is the injection to run just before executing the task injection_targets are tasks that block this task cleaning up