doot.workflow.task

A Utility implementation of most of what a task needs

Classes

_TaskActionPrep_m

_TaskProperties_m

_TaskStubbing_m

_TaskHelp_m

DootTask

The simplest task, which can import action classes.

Module Contents

class doot.workflow.task._TaskActionPrep_m[source]
prepare_actions() None[source]
if the task/action spec requires particular action ctors, load them.

if the action spec doesn’t have a ctor, use the task’s action_ctor

collects any action errors together, then raises them as a task error

Return type:

None

class doot.workflow.task._TaskProperties_m[source]
classmethod param_specs() list[jgdv.cli.ParamSpec][source]

make class parameter specs

Return type:

list[jgdv.cli.ParamSpec]

property name: doot.workflow.structs.TaskName[source]
Return type:

doot.workflow.structs.TaskName

property short_doc: str[source]

Generate Job Class 1 line help string

Return type:

str

property doc: list[str][source]
Return type:

list[str]

property is_stale: bool[source]
Return type:

bool

class doot.workflow.task._TaskStubbing_m[source]
classmethod stub_class(
stub: doot.cmds.structs.task_stub.TaskStub,
) doot.cmds.structs.task_stub.TaskStub[source]

Create a basic toml stub for this task

Parameters:

stub (doot.cmds.structs.task_stub.TaskStub)

Return type:

doot.cmds.structs.task_stub.TaskStub

stub_instance(
stub: doot.cmds.structs.task_stub.TaskStub,
) doot.cmds.structs.task_stub.TaskStub[source]

extend the class toml stub with details from this instance

Parameters:

stub (doot.cmds.structs.task_stub.TaskStub)

Return type:

doot.cmds.structs.task_stub.TaskStub

class doot.workflow.task._TaskHelp_m[source]
classmethod class_help() list[str][source]

Task class help.

Return type:

list[str]

class doot.workflow.task.DootTask(
spec: doot.workflow._interface.TaskSpec_i,
*,
action_ctor: jgdv.Maybe[collections.abc.Callable] = None,
**kwargs: Any,
)[source]

The simplest task, which can import action classes. eg: actions = [ {do = “doot.workflow.actions.shell_action:DootShellAction”, args = [“echo”, “this is a test”] } ]

Actions are imported upon task creation.

Parameters:
Flags: ClassVar[type[doot.workflow._interface.TaskMeta_e]]
INITIAL_STATE: ClassVar[doot.workflow._interface.TaskStatus_e]
COMPLETE_STATES: ClassVar[set[doot.workflow._interface.TaskStatus_e]]
_default_flags: ClassVar
action_ctor: type
_help: tuple[str, Ellipsis] = ('The Simplest Task',)
_version: str = '0.1'
_internal_state: dict
flags
_spec
_priority
_status
property name: doot.workflow._interface.TaskName_p[source]
Return type:

doot.workflow._interface.TaskName_p

property spec: doot.workflow._interface.TaskSpec_i[source]
Return type:

doot.workflow._interface.TaskSpec_i

property status: doot.workflow._interface.TaskStatus_e[source]
Return type:

doot.workflow._interface.TaskStatus_e

property priority: int[source]
Return type:

int

property internal_state: dict[source]
Return type:

dict

prepare_actions() None[source]
if the task/action spec requires particular action ctors, load them.

if the action spec doesn’t have a ctor, use the task’s action_ctor

collects any action errors together, then raises them as a task error

Return type:

None

log(
msg: str | jgdv.Lambda | list,
level: int = logmod.DEBUG,
prefix: jgdv.Maybe[str] = None,
) None[source]

utility method to log a message, useful as tasks are running

Parameters:
  • msg (str | jgdv.Lambda | list)

  • level (int)

  • prefix (jgdv.Maybe[str])

Return type:

None

get_action_group(
group_name: str,
) list[doot.workflow._interface.ActionSpec_i][source]
Parameters:

group_name (str)

Return type:

list[doot.workflow._interface.ActionSpec_i]