httk.workflow.languages.cwl

Run one CWL document as one httk₂ job.

The Common Workflow Language is supported here as a workflow language, not as an execution engine. A document is parsed and validated with cwl-utils, normalized into one self-contained JSON plan, and then executed entirely by httk₂’s own packaged cwl_runner.py on httk₂’s own manager: claims, leases, attempts, checkpoints, labeled children and journalled state frames are the same ones every other job of a workspace gets. cwltool is not used, not bundled and never invoked.

from httk.workflow import Workspace, new_job

workspace = Workspace.initialize("workflow-workspace")
job = new_job(workspace, "flow.cwl", inputs={"message": "hello"})
print(job.job_key)

Parsing needs the optional extra:

pip install httk-workflow[cwl]

which brings cwl-utils and cwl-upgrader. Nothing else in httk-workflow depends on either, and the packaged runner that executes the normalized plan needs neither: the plan is plain JSON, so the machine that runs an imported CWL job does not need a CWL library at all.

The supported subset

Supported: Workflow and CommandLineTool of CWL v1.0, v1.1 and v1.2 (older versions are upgraded when cwl-upgrader is installed); baseCommand and arguments; input bindings with position, prefix, separate and itemSeparator; the types File, Directory, string, int, long, float, double, boolean, Any, arrays of those, and optional (?) forms of all of them; stdout/stderr shortcuts and stdout/stderr redirection; output collection through outputBinding.glob with loadContents; EnvVarRequirement, ResourceRequirement (recorded), ToolTimeLimit (honoured), successCodes; workflow steps with source, default, linkMerge and plain-reference valueFrom; single-input scatter and dotproduct scatter; subworkflows; and when written as one plain parameter reference.

Rejected, always with the feature name and where it was found: any JavaScript (InlineJavascriptRequirement, ${...} bodies, and anything inside $(...) beyond a plain inputs.x/runtime.x reference), ExpressionTool and Operation, nested_crossproduct and flat_crossproduct, streamable inputs and outputs, secondaryFiles, outputEval, ShellCommandRequirement, SchemaDefRequirement, InitialWorkDirRequirement, stdin redirection, record and enum schemas, and CWL v1.2 loops.

DockerRequirement is neither rejected nor honoured: it is recorded as the required capability docker on the job — so only a manager declaring that capability will claim it — and reported as a warning, because httk₂ runs the command directly and does not pull, build or enter the image.

Attributes

FILES_DIRECTORY
PACKAGE
RUNNER
DOCUMENT_FILE
INPUTS_FILE
STAGED_DIRECTORY
DOCKER_CAPABILITY
LANGUAGE

Exceptions

LanguageOutputsMissingError	A language job has no readable published outputs document.
UnsupportedCwlError	One CWL feature outside the subset httk₂ executes.
CwlImportError	A CWL document that cannot be read, parsed or staged at all.

Classes

LanguagePorts	The named input and output ports of one language document.
LanguageRequest	The data supplied when preparing one language workflow.
LanguageScaffold	The files, runner, and hooks prepared for one language workflow.
WorkflowLanguage	The operations a workflow language exposes to the common layer.
CwlNotes	What one normalization pass accumulates besides the plan itself.

Functions

runner_reference(package, name)	Return the runner member of a job.json running one packaged runner.
payload_relative(name)	Return where one staged file lands inside a payload.
load_cwl_plan(workflow_path)	Parse, check and normalize one CWL document into the plan httk executes.
collect(record)	Convert one CWL runner output document into provenance-capable records.
load_cwl_inputs(inputs_path)	Read one CWL input object, as YAML when a YAML reader is installed.
stage_cwl_inputs(values, base, documents, files)	Return the input object with every File and Directory staged in the payload.

Package Contents

exception httk.workflow.languages.cwl.LanguageOutputsMissingError

Bases: ValueError

A language job has no readable published outputs document.

class httk.workflow.languages.cwl.LanguagePorts

The named input and output ports of one language document.

Parameters:

• inputs – Names of the document’s input ports.

• outputs – Names of the document’s output ports.

inputs: tuple[str, Ellipsis]

outputs: tuple[str, Ellipsis]

class httk.workflow.languages.cwl.LanguageRequest

The data supplied when preparing one language workflow.

Parameters:

• workflow_id – Identify the workflow being prepared.

• directory – Locate the workflow package, when it has one.

• document – Locate the source workflow document, when it has one.

• runner_options – Supply options for the language runner.

• inputs – Describe the workflow inputs.

• outputs – Describe the requested workflow outputs.

• parameters – Describe the declared workflow parameters.

• environment – Describe the declared workflow environment.

• excluded_members – Package members the realization must not stage.

workflow_id: str

directory: pathlib.Path | None

document: pathlib.Path | None

runner_options: collections.abc.Mapping[str, object]

inputs: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

outputs: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

parameters: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

environment: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

excluded_members: tuple[str, Ellipsis] = ()

class httk.workflow.languages.cwl.LanguageScaffold

The files, runner, and hooks prepared for one language workflow.

Parameters:

• documents – Text or byte documents to write into the payload.

• files – Files to stage into the payload.

• parameters – Job parameters produced by preparation.

• runner – Runner description for the job, when one is supplied.

• runner_executor – Select the runner executor.

• payload_runner – Name a runner staged in the payload.

• workdir_path – Name the workdir below the job payload.

• required_capabilities – Require these manager capabilities.

• reserved_parameters – Names reserved for per-job realization output.

• warnings – Preserve preparation warnings.

• instantiate – Supply the per-job hook called after input staging.

• finalize – Transform the per-job JobSpec immediately before its payload is prepared.

documents: collections.abc.Mapping[str, str | bytes]

files: collections.abc.Mapping[str, pathlib.Path]

parameters: collections.abc.Mapping[str, object]

runner: collections.abc.Mapping[str, object] | None

runner_executor: str = 'path'

payload_runner: str | None = None

workdir_path: str | None = None

required_capabilities: tuple[str, Ellipsis] = ()

reserved_parameters: tuple[str, Ellipsis] = ()

warnings: tuple[str, Ellipsis] = ()

instantiate: collections.abc.Callable[[httk.workflow.scaffold.InstantiateContext], object] | None = None

finalize: collections.abc.Callable[[httk.workflow.runtime_builders.JobSpec], httk.workflow.runtime_builders.JobSpec] | None = None

class httk.workflow.languages.cwl.WorkflowLanguage

The operations a workflow language exposes to the common layer.

Parameters:

• name – Name the language.

• steps – Declare the runner’s steps.

• initial_step – Select the runner’s initial step.

• matches – Identify documents belonging to the language.

• ports – Read the input and output ports of a document.

• validate_runner – Validate runner options for a document.

• prepare – Prepare a language request for execution.

• collect – Convert a completed job record into language outputs.

• document_policy – State whether package manifests require, allow, or forbid a source document.

• open_ports – Skip manifest port validation when document ports cannot be enumerated statically.

• has_default_collector – Provide a default collector path.

• allows_modes – Permit manifest data and workdir mode overrides.

• environment – Declare language-provided environment metadata.

• required_modules – Name the importable modules a job of this language needs at run time.

name: str

steps: tuple[str, Ellipsis]

initial_step: str

matches: collections.abc.Callable[[pathlib.Path], bool]

ports: collections.abc.Callable[[pathlib.Path], LanguagePorts]

validate_runner: collections.abc.Callable[[collections.abc.Mapping[str, object], pathlib.Path], None]

prepare: collections.abc.Callable[[LanguageRequest], LanguageScaffold]

collect: collections.abc.Callable[[httk.workflow.collecting.JobRecord], collections.abc.Mapping[str, object]]

document_policy: DocumentPolicy = 'required'

open_ports: bool = False

has_default_collector: bool = True

allows_modes: bool = True

environment: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

required_modules: tuple[str, Ellipsis] = ()

httk.workflow.languages.cwl.runner_reference(package, name)

Return the runner member of a job.json running one packaged runner.

The reserved pkg: form names the runner inside its own consumer package, and the digest is taken from the installed bytes, which is exactly what the manager verifies before it stages and executes them.

httk.workflow.languages.cwl.FILES_DIRECTORY = 'files'

httk.workflow.languages.cwl.payload_relative(name)

Return where one staged file lands inside a payload.

A bare name lands in FILES_DIRECTORY, so POSCAR becomes files/POSCAR — where the packaged runners read it — and a name that carries a directory of its own is used exactly as it is written.

Parameters:

name (str) – Name the staged file within the payload.

Returns:

The validated payload-relative destination.

Raises:

• httk.workflow.errors.FormatError – If the name uses a reserved payload member.

• ValueError – If the name is empty or absolute.

Return type:

pathlib.PurePosixPath

httk.workflow.languages.cwl.PACKAGE = 'httk.workflow.languages.cwl'

httk.workflow.languages.cwl.RUNNER = 'cwl_runner.py'

httk.workflow.languages.cwl.DOCUMENT_FILE = 'files/workflow.cwl.json'

httk.workflow.languages.cwl.INPUTS_FILE = 'files/inputs.json'

httk.workflow.languages.cwl.STAGED_DIRECTORY = 'files/inputs'

httk.workflow.languages.cwl.DOCKER_CAPABILITY = 'docker'

exception httk.workflow.languages.cwl.UnsupportedCwlError

Bases: ValueError

One CWL feature outside the subset httk₂ executes.

exception httk.workflow.languages.cwl.CwlImportError

Bases: ValueError

A CWL document that cannot be read, parsed or staged at all.

class httk.workflow.languages.cwl.CwlNotes

What one normalization pass accumulates besides the plan itself.

A note is never a refusal: everything here was accepted, and every entry is something an operator should nevertheless be told — a container that will not be entered, a hint that was dropped — or something the job must carry, like the capabilities a document implies.

Parameters:

• warnings – Collect accepted caveats in discovery order.

• capabilities – Collect manager capabilities required by the document.

warnings: list[str] = []

capabilities: set[str]

httk.workflow.languages.cwl.load_cwl_plan(workflow_path)

Parse, check and normalize one CWL document into the plan httk executes.

Parameters:

workflow_path (str | os.PathLike[str]) – Read the CWL document at this path.

Returns:

The normalized plan and accepted import notes.

Raises:

• httk.workflow.languages.cwl.CwlImportError – If the document cannot be read or parsed.

• OSError – If accessing the workflow path fails while reading its declared version.

• UnicodeError – If the workflow path is not valid UTF-8 while reading its declared version.

• httk.workflow.languages.cwl.UnsupportedCwlError – If the document uses an unsupported CWL feature.

Return type:

tuple[dict[str, object], CwlNotes]

httk.workflow.languages.cwl.collect(record)

Convert one CWL runner output document into provenance-capable records.

httk.workflow.languages.cwl.LANGUAGE

httk.workflow.languages.cwl.load_cwl_inputs(inputs_path)

Read one CWL input object, as YAML when a YAML reader is installed.

Parameters:

inputs_path (str | os.PathLike[str]) – Read the CWL input object at this path.

Returns:

The input values keyed by CWL input name.

Raises:

• httk.workflow.languages.cwl.CwlImportError – If the input object cannot be read or is not a mapping.

• OSError – If the input path cannot be read.

• UnicodeError – If the input path is not valid UTF-8.

Return type:

dict[str, object]

httk.workflow.languages.cwl.stage_cwl_inputs(values, base, documents, files)

Return the input object with every File and Directory staged in the payload.

A staged entry carries a payload-relative payload member instead of a path, so the job is self-contained: it travels to another machine with the files its first tool reads.

Parameters:

• values (collections.abc.Mapping[str, object]) – Stage these decoded CWL input values.

• base (pathlib.Path) – Resolve relative input paths from this directory.

• documents (dict[str, str]) – Add literal files and normalized documents to this mapping.

• files (dict[str, str | os.PathLike[str]]) – Add source files to this mapping for payload staging.

Returns:

The input values rewritten with payload-relative paths.

Raises:

• httk.workflow.languages.cwl.CwlImportError – If a local input path is missing or invalid.

• httk.workflow.languages.cwl.UnsupportedCwlError – If an input uses an unsupported CWL feature.

Return type:

dict[str, object]