speedwagon

Speedwagon.

Functions

speedwagon.available_workflows() → dict

Locate all workflow class in workflows subpackage.

This looks for a workflow prefix in the naming.

Returns:

Dictionary of all workflow

Classes

class speedwagon.Workflow(*args, **kwargs)

Base class for defining a new workflow item.

Subclass this class to generate a new workflow.

Notes

You need to implement the discover_task_metadata() method.

__init__(*args, **kwargs) → None

Populate the base structure of a workflow class.

completion_task(task_builder: TaskBuilder, results, **user_args) → None

Last task after Job is completed.

create_new_task(task_builder: TaskBuilder, **job_args) → None

Add a new task to be accomplished when the workflow is started.

Use the task_builder parameter’s add_subtask method to include a speedwagon.Subtask()

Example:

title_page = job_args['title_page']
source = job_args['source_path']
package_id = job_args['package_id']

task_builder.add_subtask(
    subtask=MakeYamlTask(package_id, source, title_page))

abstract discover_task_metadata(initial_results: List[Any], additional_data: Dict[str, Any], **user_args) → List[dict]

Generate data or parameters needed for upcoming tasks.

Generate data or parameters needed for task to complete based on the user’s original configuration

Return a list of dictionaries of types that can be serialized,

preferably strings.

classmethod generate_report(results: List[Result], **user_args) → Optional[str]

Generate a text report for the results of the workflow.

Example

report_lines = []

for checksum_report, items_written in \\
        cls.sort_results([i.data for \n
        i in results]).items():

    report_lines.append(
        f"Checksum values for {len(items_written)} "
        f"files written to {checksum_report}")

return '\\n'.join(report_lines)

get_additional_info(user_request_factory: UserRequestFactory, options: dict, pretask_results: list) → dict

Request additional information from the user.

If a user needs to be prompted for more information, run this

Parameters:

• user_request_factory – factory needed

• options – Dictionary of existing user settings

• pretask_results – results of the pretask, if any

Returns:

Any additional configurations that needs to be added to a job

get_user_options() → List[AbsOutputOptionDataType]

Get user options.

Defaults to no args.

initial_task(task_builder: TaskBuilder, **user_args) → None

Create a task to run before the main tasks start.

The initial task is run prior to the get_additional_info method. Results generated here will then be passed to get_additional_info.

This is useful for locating additional information that will be needed by other tasks and the user needs to additional decisions before running the main tasks.

In general, prefer discover_task_metadata() if you don’t need a user’s interaction.

static validate_user_options(**user_args) → bool

Make sure that the options the user provided is valid.

Notes

This raises a ValueError on Failure. This should be rewritten to be better.

Parameters:

**user_args –

Returns:

Returns True on valid else returns False.