m3resp.workflows.WorkflowService

class m3resp.workflows.WorkflowService

Discover workflow steps, inspect specs and run workflows for an application.

Methods

compile_workflow	Load and compile a workflow into an ordered description of its steps.
describe_capability	Return one step's settings, input/output descriptions and availability.
list_capabilities	Return registered step descriptions, optionally filtered by name prefix.
run_workflow	Run a workflow synchronously and return an execution summary.
validate_workflow	Load a workflow and return its structural and optional readiness report.

compile_workflow(spec)

Load and compile a workflow into an ordered description of its steps.

Parameters:

spec (str | Path | dict[str, Any] | WorkflowSpec) – A WorkflowSpec, dictionary, or YAML/JSON file path.

Returns:

Workflow name, schema version and compiled steps with

resolved input/output names, defaults, parameter values and paths.

Return type:

dict[str, Any]

Raises:

• WorkflowSpecError – If parsing or structural validation fails.

• UnknownStepError – If the first structural error names an unknown step.

• OSError – If a spec file cannot be read.

describe_capability(operation_id)

Return one step’s settings, input/output descriptions and availability.

Raises UnknownStepError if operation_id is unregistered. Former step names resolve through registered aliases.

Parameters:

operation_id (str)

Return type:

dict[str, Any]

list_capabilities(*, prefix=None)

Return registered step descriptions, optionally filtered by name prefix.

Each dictionary includes parameters, input/output descriptions and current availability. Names are sorted; prefix can select e.g. "eit." steps.

Parameters:

prefix (str | None)

Return type:

list[dict[str, Any]]

run_workflow(spec, *, event_sink=None, cancellation_token=None)

Run a workflow synchronously and return an execution summary.

Parameters:

• spec (str | Path | dict[str, Any] | WorkflowSpec) – A WorkflowSpec, dictionary, or YAML/JSON file path.

• event_sink (Callable[[dict[str, Any]], None] | None) – Optional callable receiving progress-event dictionaries.

• cancellation_token (CancellationToken | None) – Optional flag checked before and after each step.

Returns:

Run identifiers, status, timings, step records, diagnostics,

warnings and output summaries. Completed work is preserved when cancelled. Output data are represented by scalar values or brief type/shape/length descriptions, as in summarize_workflow_result.

Return type:

dict[str, Any]

Raises:

• WorkflowSpecError – If the spec, bindings or returned outputs are invalid.

• UnknownStepError – If the first structural error names an unknown step.

• WorkflowExecutionError – If a step function raises, with its cause and gathered records attached.

• OSError – If a spec file cannot be read.

validate_workflow(spec, *, readiness=False)

Load a workflow and return its structural and optional readiness report.

Parameters:

• spec (str | Path | dict[str, Any] | WorkflowSpec) – A WorkflowSpec, dictionary, or YAML/JSON file path.

• readiness (bool) – Also check package availability and input-file existence.

Returns:

is_valid plus structural and readiness diagnostic lists.

Structural problems in a parsed spec are returned as diagnostics. is_valid reflects structural errors only.

Return type:

dict[str, Any]

Raises:

• WorkflowSpecError – If the spec cannot be parsed or fails format validation.

• OSError – If a spec file cannot be read.