dootle.control.fsm.task

An FSM b acked Task and job

Classes

_Predicates_m
_Callbacks_m
FSMTask	The implementation of a task, as the domain model for a TaskMachine
FSMJob	Extends an FSMTask for running a job

Module Contents

class dootle.control.fsm.task._Predicates_m

spec: doot.workflow.TaskSpec

name: doot.workflow._interface.TaskName_p

priority: int

spec_missing(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → bool

cancels the task if the spec is not registered

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

bool

should_disable(

source: statemachine.State,

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → bool

cancels the task if the spec is disabled

Parameters:

• source (statemachine.State)

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

bool

should_timeout() → bool

Cancel if you’ve waited too long

Return type:

bool

should_wait(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → bool

if any dependencies have not run, delay this task

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

bool

should_skip(source: statemachine.State) → bool

run a task’s depends_on group, coercing to a bool returns False if the runner should skip the rest of the task

Parameters:

source (statemachine.State)

Return type:

bool

should_halt(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → bool

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

bool

should_fail() → bool

Return type:

bool

state_is_needed(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → bool

delays exit from teardown _internal_state until it is safe to do so

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

bool

class dootle.control.fsm.task._Callbacks_m

_internal_state: dict

name: doot.workflow._interface.TaskName_p

spec: doot.workflow.TaskSpec

_state_history: list

on_exit_state(*, source: Any) → None

Keep track of the progression of the task

Parameters:

source (Any)

Return type:

None

on_enter_INIT(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

parent: jgdv.Maybe[doot.workflow._interface.TaskName_p] = None,

) → None

initialise _internal_state, possibly run injections?

Parameters:

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

• parent (jgdv.Maybe[doot.workflow._interface.TaskName_p])

Return type:

None

on_enter_RUNNING(

*,

step: int,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

• step (int)

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_exit_RUNNING(

*,

step: int,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

• step (int)

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_exit_TEARDOWN(

*,

source: Any,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

• source (Any)

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_enter_SUCCESS(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_enter_FAILED(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_enter_HALTED(

*,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

on_enter_SKIPPED() → None

Return type:

None

_get_parent_data(

tracker: doot.control.tracker._interface.WorkflowTracker_p,

parent: jgdv.Maybe[doot.workflow._interface.TaskName_p],

) → jgdv.Maybe[dict]

Parameters:

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

• parent (jgdv.Maybe[doot.workflow._interface.TaskName_p])

Return type:

jgdv.Maybe[dict]

_get_cli_data(

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → jgdv.Maybe[dict]

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

jgdv.Maybe[dict]

_get_spec_data() → dict

Return type:

dict

_get_inject_data(

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → jgdv.Maybe[dict]

Parameters:

tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

jgdv.Maybe[dict]

class dootle.control.fsm.task.FSMTask(spec: doot.workflow.TaskSpec)

The implementation of a task, as the domain model for a TaskMachine

Parameters:

spec (doot.workflow.TaskSpec)

_default_flags: ClassVar[set]

step: int

spec: doot.workflow.TaskSpec

status: doot.workflow._interface.TaskStatus_e

priority: int

records: list[Any]

_internal_state: dict

_state_history: list[doot.workflow._interface.TaskStatus_e]

property name: doot.workflow.TaskName

Return type:

doot.workflow.TaskName

property internal_state: dict

Return type:

dict

_execute_action_group(

*,

group: str,

lock_state: bool = False,

) → tuple[int, doot.workflow._interface.ActionResponse_e]

Execute a group of actions, possibly queue any task specs they produced, and return a count of the actions run + the result

Parameters:

• group (str)

• lock_state (bool)

Return type:

tuple[int, doot.workflow._interface.ActionResponse_e]

_execute_action(

count: int,

action: doot.workflow.ActionSpec,

*,

group: jgdv.Maybe[str] = None,

lock_state: bool = False,

) → doot.workflow._interface.ActionResponse_e | bool | list[doot.workflow.TaskSpec]

Run the given action of a specific task.

returns either a list of specs to (potentially) queue, or an ActRE describing the action result.

Parameters:

• count (int)

• action (doot.workflow.ActionSpec)

• group (jgdv.Maybe[str])

• lock_state (bool)

Return type:

doot.workflow._interface.ActionResponse_e | bool | list[doot.workflow.TaskSpec]

get_action_group(

group_name: str,

) → list[doot.workflow.ActionSpec]

Parameters:

group_name (str)

Return type:

list[doot.workflow.ActionSpec]

param_specs() → list

Return type:

list

log(

msg: str,

level: int = logmod.DEBUG,

prefix: jgdv.Maybe[str] = None,

) → None

Parameters:

• msg (str)

• level (int)

• prefix (jgdv.Maybe[str])

Return type:

None

class dootle.control.fsm.task.FSMJob(spec: doot.workflow.TaskSpec)

Bases: FSMTask

Extends an FSMTask for running a job

Parameters:

spec (doot.workflow.TaskSpec)

on_enter_RUNNING(

step: int,

tracker: doot.control.tracker._interface.WorkflowTracker_p,

) → None

Modifies how the object runs,

requires the main action group return a list of tasks/specs

Parameters:

• step (int)

• tracker (doot.control.tracker._interface.WorkflowTracker_p)

Return type:

None

_execute_expansion_group(

*,

group: str,

) → tuple[int, list[doot.workflow.TaskSpec]]

Execute a group of actions, possibly queue any task specs they produced, and return a count of the actions run + the result

Parameters:

group (str)

Return type:

tuple[int, list[doot.workflow.TaskSpec]]