speedwagon.tasks

Define a single step in the workflow.

class speedwagon.tasks.Subtask

Base class for defining a new task for a Workflow to create.

Subclass this generate a new task

__init__() → None

Create a new sub-task.

exec() → None

Execute subtask.

log(message: str) → None

Generate text message for the subtask.

property parent_task_log_q: Deque[str]

Log queue of the parent.

property results: _T | None

Get the results of the subtask.

set_results(results: _T) → None

Set the results of the subtask.

property settings: Dict[str, str]

Get the settings for the subtask.

property status: TaskStatus

Get the status of the subtask.

property subtask_working_dir: str

Get the subtask working directory.

Notes

This has the side effect of creating the working directory if it does not already exist.

task_description() → str | None

Get user readable information about what the subtask is doing.

property task_result: Result[_T] | None

Get the result of the subtask.

work() → bool

Perform work.

This method is called when the task’s work should be done.

Override this method to accomplish the task.

Note

Currently expects to return a boolean value to indicate if the task has succeeded or failed. However, this is likely to change.

class speedwagon.tasks.Result(source: Type[AbsSubtask], data: _T)

Subtask result.

source

Class Type used to provide the results

Type:

Type[speedwagon.tasks.tasks.AbsSubtask]

data

Payload of the result data.

Type:

speedwagon.tasks.tasks._T