speedwagon

Speedwagon.

Decorators

@speedwagon.hookimpl(function: _F | None = None, hookwrapper: bool = False, optionalhook: bool = False, tryfirst: bool = False, trylast: bool = False, specname: str | None = None, wrapper: bool = False) → _F | Callable[[_F], _F]

Decorator for marking functions as hook implementations.

Instantiate it with a project_name to get a decorator. Calling PluginManager.register() later will discover all marked functions if the PluginManager uses the same project name.

Functions

speedwagon.available_workflows(strategy: AbsWorkflowFinder | None = None) → 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

Create a new workflow object.

completion_task(task_builder: TaskBuilder, results: List[Result], user_args: _T) → None

Last task after Job is completed.

By default, this method is a no-op unless overridden.

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))

By default, this method is a no-op unless overridden.

abstract discover_task_metadata(initial_results: List[Result], additional_data: Mapping[str, Any], user_args: _T) → Sequence[Mapping[str, object]]

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: _T) → str | None

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)

By default, this method is a no-op and returns None unless overridden.

get_additional_info(user_request_factory: UserRequestFactory, options: _T, pretask_results: List[Result]) → Mapping[str, Any]

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_workflow_configuration_value(key: str) → SettingsDataType | None

Get a value from the workflow configuration.

initial_task(task_builder: TaskBuilder, user_args: _T) → 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.

By default, this method is a no-op unless overridden.

job_options() → List[AbsOutputOptionDataType]

Get user options.

Defaults to no args.

set_options_backend(value: AbsWorkflowBackend) → None

Set the option backend.

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 – user arguments

Returns:

Returns True on valid else returns False.

workflow_options() → List[AbsOutputOptionDataType]

Get options configured at the application level.

Defaults to no args.