doot.workflow.factory

Type Aliases

SpecialType

Classes

`TaskFactory`	Factory to create task specs, instantiate them, and make tasks
`SubTaskFactory`	Additional factory for generating related tasks of an instantiated spec

Module Contents

doot.workflow.factory.SpecialType: TypeAlias = _SpecialType

class doot.workflow.factory.TaskFactory( *, spec_ctor: jgdv.Maybe[type] = None, task_ctor: jgdv.Maybe[type] = None, job_ctor: jgdv.Maybe[type] = None, )[source]

Factory to create task specs, instantiate them, and make tasks

build : data -> spec delay : data -> delayed -> spec instantiate : spec -> spec(name=name[uuid]) reify : spec,partial -> spec over : orig,plus -> spec(plus<orig, name..<+>[uuid]) under : orig,plus -> spec(orig<plus, name..<+>[uuid]) make : spec -> task

Parameters:

spec_ctor (jgdv.Maybe[type])
task_ctor (jgdv.Maybe[type])
job_ctor (jgdv.Maybe[type])

spec_ctor: type[doot.workflow._interface.TaskSpec_i]

task_ctor: type[doot.workflow._interface.Task_p]

job_ctor: type[doot.workflow._interface.Job_p]

build( data: jgdv.structs.chainguard.ChainGuard | dict | doot.workflow._interface.TaskName_p | str, ) → doot.workflow._interface.TaskSpec_i[source]

Parameters:: data (jgdv.structs.chainguard.ChainGuard | dict | doot.workflow._interface.TaskName_p | str)
Return type:: doot.workflow._interface.TaskSpec_i

delay( *, base: doot.workflow._interface.TaskName_p, target: doot.workflow._interface.TaskName_p, inject: jgdv.Maybe[doot.workflow._interface.InjectSpec_i] = None, applied: jgdv.Maybe[dict] = None, overrides: dict, ) → doot.workflow._interface.DelayedSpec[source]

Build data structure that the registry will process into a full spec

Parameters:

base (doot.workflow._interface.TaskName_p)
target (doot.workflow._interface.TaskName_p)
inject (jgdv.Maybe[doot.workflow._interface.InjectSpec_i])
applied (jgdv.Maybe[dict])
overrides (dict)

Return type:

doot.workflow._interface.DelayedSpec

instantiate( obj: doot.workflow._interface.TaskSpec_i, *, suffix: jgdv.Maybe[bool | str] = None, extra: jgdv.Maybe[collections.abc.Mapping] = None, ) → doot.workflow._interface.TaskSpec_i[source]

Return this spec, copied with a uniq name

Parameters:

obj (doot.workflow._interface.TaskSpec_i)
suffix (jgdv.Maybe[bool | str])
extra (jgdv.Maybe[collections.abc.Mapping])

Return type:

doot.workflow._interface.TaskSpec_i

merge( top: dict | doot.workflow._interface.TaskSpec_i, bot: dict | doot.workflow._interface.TaskSpec_i, *, suffix: jgdv.Maybe[str | Literal[False]] = None, name: jgdv.Maybe[doot.workflow._interface.TaskName_p] = None, ) → doot.workflow._interface.TaskSpec_i[source]

bot + top -> TaskSpec

Parameters:

top (dict | doot.workflow._interface.TaskSpec_i)
bot (dict | doot.workflow._interface.TaskSpec_i)
suffix (jgdv.Maybe[str | Literal[False]])
name (jgdv.Maybe[doot.workflow._interface.TaskName_p])

Return type:

doot.workflow._interface.TaskSpec_i

make( obj: doot.workflow._interface.TaskSpec_i, ensure: Any = None, ) → doot.workflow._interface.Task_p[source]

Create actual task instance

if no spec_ctor has been specified, uses the default spec_ctor for job/task

Parameters:

obj (doot.workflow._interface.TaskSpec_i)
ensure (Any)

Return type:

doot.workflow._interface.Task_p

get_source_names( obj: doot.workflow._interface.TaskSpec_i, ) → list[doot.workflow._interface.TaskName_p][source]

Get from the spec’s sources just its source tasks

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: list[doot.workflow._interface.TaskName_p]

action_groups( obj: doot.workflow._interface.TaskSpec_i, ) → collections.abc.Iterable[collections.abc.Iterable][source]

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: collections.abc.Iterable[collections.abc.Iterable]

action_group_elements( obj: doot.workflow._interface.TaskSpec_i, ) → collections.abc.Iterable[doot.workflow._interface.ActionSpec_i | doot.workflow._interface.RelationSpec_i][source]

Get the elements of: depends_on, setup, actions, and require_for.

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: collections.abc.Iterable[doot.workflow._interface.ActionSpec_i | doot.workflow._interface.RelationSpec_i]

_specialize_merge(*, bot: dict, top: dict) → dict[source]

Apply top over the top of bot

Combines, rather than overrides, particular values.

Parameters:

bot (dict)
top (dict)

Return type:

dict

_prep_name( base: doot.workflow._interface.TaskName_p, *, suffix: jgdv.Maybe[int | str | Literal[False]] = None, ) → doot.workflow._interface.TaskName_p[source]

Parameters:

base (doot.workflow._interface.TaskName_p)
suffix (jgdv.Maybe[int | str | Literal[False]])

Return type:

doot.workflow._interface.TaskName_p

class doot.workflow.factory.SubTaskFactory[source]

Additional factory for generating related tasks of an instantiated spec

generate_names( obj: doot.workflow._interface.TaskSpec_i, ) → list[doot.workflow._interface.TaskName_p][source]

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: list[doot.workflow._interface.TaskName_p]

generate_specs( obj: doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec, ) → list[dict][source]

Parameters:: obj (doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec)
Return type:: list[dict]

_gen_job_head( obj: doot.workflow._interface.TaskSpec_i, ) → list[dict][source]

Generate a top spec for a job, taking the jobs cleanup actions and using them as the head’s main action. Cleanup relations are turning into the head’s dependencies Depends on the job, and its reactively queued.

Equivalent to: await job.depends_on() await job.setup() subtasks = job.actions() await subtasks job.head() await job.cleanup()

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: list[dict]

_gen_cleanup_task( obj: doot.workflow._interface.TaskSpec_i, ) → list[dict][source]

Generate a cleanup task, shifting the ‘cleanup’ actions and dependencies to ‘depends_on’ and ‘actions’

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: list[dict]

_raw_data_to_specs( deps: list[str | dict], *, relation: doot.workflow._interface.RelationMeta_e = DEFAULT_RELATION, ) → list[doot.workflow._interface.ActionSpec_i | doot.workflow._interface.RelationSpec_i][source]

Convert toml provided raw data (str’s, dicts) of specs into ActionSpec and RelationSpec object

Parameters:

deps (list[str | dict])
relation (doot.workflow._interface.RelationMeta_e)

Return type:

list[doot.workflow._interface.ActionSpec_i | doot.workflow._interface.RelationSpec_i]

_job_head_p( obj: doot.workflow._interface.TaskSpec_i, ) → jgdv.Maybe[doot.workflow._interface.TaskName_p][source]

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: jgdv.Maybe[doot.workflow._interface.TaskName_p]

_cleanup_p( obj: doot.workflow._interface.TaskSpec_i, ) → jgdv.Maybe[doot.workflow._interface.TaskName_p][source]

Parameters:: obj (doot.workflow._interface.TaskSpec_i)
Return type:: jgdv.Maybe[doot.workflow._interface.TaskName_p]