"""Importing one CWL document as one *httk₂* job.
The `Common Workflow Language <https://www.commonwl.org/>`_ is supported here as
a workflow **language**, not as an execution engine. A document is parsed and
validated with `cwl-utils <https://github.com/common-workflow-language/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.**
.. code-block:: python
from httk.workflow import Workspace
from httk.workflow.compat.cwl import import_cwl
workspace = Workspace.initialize("workflow-workspace")
imported = import_cwl(workspace, "flow.cwl", "job.yml", tag="echo")
print(imported.job.job_key, imported.warnings)
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.
"""
import json
import logging
import os
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from importlib import import_module
from importlib.util import find_spec
from pathlib import Path, PurePosixPath
from tempfile import TemporaryDirectory
from typing import Any, Literal
from urllib.parse import unquote, urlparse
from httk.workflow import Workspace
from .._integration import (
DEFAULT_PLACEMENT,
FILES_DIRECTORY,
ScaffoldedJob,
submit_integration_job,
)
_LOGGER = logging.getLogger(__name__)
#: The workflow name every imported CWL job carries.
[docs]
WORKFLOW = "cwl.workflow"
#: The step an imported CWL job starts at.
#: The package the packaged runner is resolved and digest-pinned within.
#: The packaged runner an imported CWL job runs.
[docs]
RUNNER = "cwl_runner.py"
#: Where the normalized plan and the input object are staged in the payload.
[docs]
DOCUMENT_FILE = f"{FILES_DIRECTORY}/workflow.cwl.json"
#: Where File and Directory inputs are staged in the payload.
[docs]
STAGED_DIRECTORY = f"{FILES_DIRECTORY}/inputs"
#: The capability a document asking for a container is filed under.
[docs]
DOCKER_CAPABILITY = "docker"
#: What ``pip install httk-workflow[cwl]`` provides, named in the error a missing
#: install produces.
_CWL_EXTRA = "pip install httk-workflow[cwl]"
#: The process classes this importer understands.
_PROCESS_CLASSES = ("Workflow", "CommandLineTool")
#: The scalar types this importer understands, CWL name to normalized name.
_SCALAR_TYPES = {
"string": "string",
"int": "int",
"long": "int",
"float": "float",
"double": "float",
"boolean": "boolean",
"File": "File",
"Directory": "Directory",
"Any": "Any",
"stdout": "stdout",
"stderr": "stderr",
}
#: Requirements and hints that change nothing here and are carried anyway.
_RECORDED_REQUIREMENTS = (
"ResourceRequirement",
"SoftwareRequirement",
"NetworkAccess",
"LoadListingRequirement",
"WorkReuse",
"ScatterFeatureRequirement",
"SubworkflowFeatureRequirement",
"MultipleInputFeatureRequirement",
"StepInputExpressionRequirement",
"ToolTimeLimit",
"EnvVarRequirement",
)
#: Requirements this importer refuses, and why each one is outside the subset.
_REFUSED_REQUIREMENTS = {
"InlineJavascriptRequirement": "httk runs no JavaScript engine, so only plain parameter references are supported",
"ShellCommandRequirement": "httk runs an argument vector, never a shell command line",
"SchemaDefRequirement": "record and enum schemas are outside the supported type set",
"InitialWorkDirRequirement": "httk stages File inputs itself and does not build a working directory from a listing",
"InplaceUpdateRequirement": "httk never lets a tool write to its input files",
}
#: One plain parameter reference: a namespace, a name, and dotted field access.
_PLAIN_REFERENCE = re.compile(r"\$\((inputs|runtime)\.[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*\)")
#: Anything at all that CWL would evaluate.
_ANY_EXPRESSION = re.compile(r"\$[({]")
[docs]
class UnsupportedCwlError(ValueError):
"""One CWL feature outside the subset *httk₂* executes."""
[docs]
class CwlImportError(ValueError):
"""A CWL document that cannot be read, parsed or staged at all."""
@dataclass
[docs]
class 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.
"""
#: Everything accepted with a caveat, in the order it was found.
[docs]
warnings: list[str] = field(default_factory=list)
#: The capabilities the imported job must require of a manager.
[docs]
capabilities: set[str] = field(default_factory=set)
@dataclass(frozen=True)
[docs]
class ImportedCwl:
"""One imported CWL document: the job, the plan, and what was noted."""
#: The submitted job.
#: The normalized plan staged in its payload.
[docs]
document: Mapping[str, object]
#: The input object staged in its payload.
#: Everything the import accepted but wants an operator to know about.
[docs]
warnings: tuple[str, ...]
# ---------------------------------------------------------------------------
# Expressions, identifiers and types
# ---------------------------------------------------------------------------
def _name(identifier: object) -> str:
"""Return the short name of one absolute CWL identifier."""
text = str(identifier)
fragment = text.partition("#")[2] or text
return fragment.rsplit("/", 1)[-1]
def _source_name(identifier: object, steps: Sequence[str] = ()) -> str:
"""Return one output source as ``step/port``, or as a workflow input name.
Identifiers are absolute and nested — an inlined subworkflow's own input is
spelled ``…#outer/run/message`` — so the last component alone is ambiguous
and the last *two* would read a nesting level as a step. The step names of
the workflow being normalized are what tells the two apart.
"""
text = str(identifier)
fragment = text.partition("#")[2] or PurePosixPath(text).name
parts = fragment.split("/")
if len(parts) > 1 and parts[-2] in set(steps):
return f"{parts[-2]}/{parts[-1]}"
return parts[-1]
def _check_expression(value: object, where: str) -> object:
"""Refuse anything CWL would evaluate beyond a plain parameter reference."""
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
for item in value:
_check_expression(item, where)
return value
if not isinstance(value, str) or not _ANY_EXPRESSION.search(value):
return value
if "${" in value:
raise UnsupportedCwlError(
f"unsupported CWL feature JavaScript expression at {where}: {value!r}; "
"httk evaluates plain parameter references such as $(inputs.name) and nothing else"
)
remaining = _PLAIN_REFERENCE.sub("", value)
if _ANY_EXPRESSION.search(remaining):
raise UnsupportedCwlError(
f"unsupported CWL feature expression at {where}: {value!r}; "
"httk evaluates plain parameter references such as $(inputs.name) or $(runtime.outdir) only"
)
return value
def _normalize_type(value: object, where: str) -> dict[str, object]:
"""Normalize one CWL type into ``{type, items?, optional}``."""
optional = False
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) and not isinstance(value, Mapping):
members = [item for item in value if item != "null"]
optional = len(members) != len(list(value))
if len(members) != 1:
raise UnsupportedCwlError(
f"unsupported CWL feature union type at {where}: {value!r}; httk supports one type, optionally null"
)
normalized = _normalize_type(members[0], where)
normalized["optional"] = True
return normalized
if isinstance(value, Mapping):
kind = value.get("type")
if kind == "array":
items = _normalize_type(value.get("items"), f"{where} items")
return {"type": "array", "items": items, "optional": optional}
raise UnsupportedCwlError(
f"unsupported CWL feature {kind!r} schema at {where}; "
"httk supports File, Directory, string, int, long, float, double, boolean, Any and arrays of them"
)
text = str(value)
if text.endswith("?"):
normalized = _normalize_type(text[:-1], where)
normalized["optional"] = True
return normalized
if text.endswith("[]"):
return {"type": "array", "items": _normalize_type(text[:-2], where), "optional": optional}
if text not in _SCALAR_TYPES:
raise UnsupportedCwlError(
f"unsupported CWL feature type {text!r} at {where}; "
"httk supports File, Directory, string, int, long, float, double, boolean, Any and arrays of them"
)
return {"type": _SCALAR_TYPES[text], "optional": optional}
def _binding(value: object, where: str) -> dict[str, object] | None:
"""Normalize one ``inputBinding``, refusing what cannot be built into argv."""
if value is None:
return None
if not isinstance(value, Mapping):
raise CwlImportError(f"the inputBinding at {where} is not an object")
if value.get("shellQuote") is False:
raise UnsupportedCwlError(
f"unsupported CWL feature shellQuote at {where}; httk runs an argument vector, never a shell command line"
)
if value.get("loadContents"):
raise UnsupportedCwlError(
f"unsupported CWL feature loadContents on an inputBinding at {where}; "
"httk loads contents on output bindings only"
)
binding: dict[str, object] = {}
position = value.get("position")
if position is not None:
_check_expression(position, f"{where} position")
if isinstance(position, bool) or not isinstance(position, int):
raise UnsupportedCwlError(
f"unsupported CWL feature computed position at {where}: {position!r}; httk sorts by literal positions"
)
binding["position"] = position
for member in ("prefix", "itemSeparator", "valueFrom"):
item = value.get(member)
if item is not None:
binding[member] = _check_expression(item, f"{where} {member}")
if value.get("separate") is not None:
binding["separate"] = bool(value["separate"])
return binding
# ---------------------------------------------------------------------------
# Requirements
# ---------------------------------------------------------------------------
def _requirements(process: Mapping[str, object], where: str, context: CwlNotes) -> dict[str, object]:
"""Normalize the requirements and hints of one process, refusing the rest."""
collected: dict[str, object] = {}
for member, mandatory in (("hints", False), ("requirements", True)):
entries = process.get(member) or []
if isinstance(entries, Mapping):
entries = [{"class": name, **(body if isinstance(body, Mapping) else {})} for name, body in entries.items()]
if not isinstance(entries, Sequence):
raise CwlImportError(f"the {member} of {where} are neither an array nor an object")
for entry in entries:
if not isinstance(entry, Mapping):
raise CwlImportError(f"one of the {member} of {where} is not an object")
name = str(entry.get("class", "")).rsplit("#", 1)[-1]
if name in _REFUSED_REQUIREMENTS:
if not mandatory:
# A hint is advisory by definition, so an unsupported one is
# dropped with a warning rather than refused.
context.warnings.append(f"{where}: the {name} hint is ignored")
continue
raise UnsupportedCwlError(f"unsupported CWL feature {name} at {where}: {_REFUSED_REQUIREMENTS[name]}")
if name == "DockerRequirement":
context.capabilities.add(DOCKER_CAPABILITY)
context.warnings.append(
f"{where}: DockerRequirement is recorded as the required capability {DOCKER_CAPABILITY!r} "
"and is otherwise ignored; httk runs the command directly and never pulls or enters an image"
)
collected[name] = {key: item for key, item in entry.items() if key != "class"}
continue
if name not in _RECORDED_REQUIREMENTS:
if not mandatory:
context.warnings.append(f"{where}: the {name} hint is ignored")
continue
raise UnsupportedCwlError(
f"unsupported CWL feature {name} at {where}; it is not part of the CWL subset httk executes"
)
body = {key: _check_expression(item, f"{where} {name}") for key, item in entry.items() if key != "class"}
collected[name] = body
return collected
# ---------------------------------------------------------------------------
# Processes
# ---------------------------------------------------------------------------
def _normalize_process(saved: Mapping[str, object], where: str, context: CwlNotes) -> dict[str, object]:
"""Normalize one saved CWL process into the plan the runner executes."""
process_class = str(saved.get("class", "")).rsplit("#", 1)[-1]
if process_class not in _PROCESS_CLASSES:
raise UnsupportedCwlError(
f"unsupported CWL feature class {process_class or '(none)'} at {where}; "
f"httk executes {' and '.join(_PROCESS_CLASSES)} documents"
)
if saved.get("loop") is not None:
raise UnsupportedCwlError(f"unsupported CWL feature loop at {where}; httk has no CWL loop support")
plan: dict[str, object] = {
"class": process_class,
"id": _name(saved.get("id", where)),
"requirements": _requirements(saved, where, context),
"inputs": _normalize_inputs(saved, where, context),
}
if process_class == "CommandLineTool":
plan.update(_normalize_tool(saved, where, context))
else:
plan.update(_normalize_workflow(saved, where, context))
return plan
def _normalize_inputs(saved: Mapping[str, object], where: str, context: CwlNotes) -> dict[str, object]:
"""Normalize the input parameters of one process."""
inputs: dict[str, object] = {}
for entry in _entries(saved.get("inputs"), f"{where} inputs"):
name = _name(entry.get("id", ""))
location = f"{where}.{name}"
_reject_parameter_extras(entry, location)
parameter: dict[str, object] = {"type": _normalize_type(entry.get("type"), location)}
if "default" in entry and entry["default"] is not None:
parameter["default"] = _check_expression(entry["default"], f"{location} default")
binding = _binding(entry.get("inputBinding"), f"{location} inputBinding")
if binding is not None:
parameter["inputBinding"] = binding
inputs[name] = parameter
return inputs
def _reject_parameter_extras(entry: Mapping[str, object], location: str) -> None:
"""Refuse the parameter members outside the subset."""
if entry.get("secondaryFiles"):
raise UnsupportedCwlError(
f"unsupported CWL feature secondaryFiles at {location}; httk stages exactly the files an input names"
)
if entry.get("streamable"):
raise UnsupportedCwlError(
f"unsupported CWL feature streamable at {location}; httk stages whole files and runs no pipes between tools"
)
def _normalize_tool(saved: Mapping[str, object], where: str, context: CwlNotes) -> dict[str, object]:
"""Normalize the parts that only a ``CommandLineTool`` has."""
base = saved.get("baseCommand")
if base is None:
command: list[str] = []
elif isinstance(base, str):
command = [base]
elif isinstance(base, Sequence):
command = [str(item) for item in base]
else:
raise CwlImportError(f"the baseCommand of {where} is neither a string nor an array")
arguments: list[dict[str, object]] = []
for index, entry in enumerate(_as_list(saved.get("arguments"))):
location = f"{where} argument {index}"
if isinstance(entry, str):
arguments.append({"valueFrom": _check_expression(entry, location), "position": 0})
continue
if not isinstance(entry, Mapping):
raise CwlImportError(f"{location} is neither a string nor a binding")
binding = _binding(entry, location) or {}
binding.setdefault("position", 0)
arguments.append(binding)
tool: dict[str, object] = {
"baseCommand": command,
"arguments": arguments,
"outputs": _normalize_tool_outputs(saved, where),
}
if saved.get("stdin") is not None:
raise UnsupportedCwlError(
f"unsupported CWL feature stdin redirection at {where}; httk runs an argument vector without a shell, "
"so a tool reading a file must be given it as an argument"
)
for member in ("stdout", "stderr"):
value = saved.get(member)
if value is not None:
tool[member] = _check_expression(value, f"{where} {member}")
codes = saved.get("successCodes")
if codes:
tool["successCodes"] = [int(str(code)) for code in _as_list(codes)]
if not command and not arguments:
raise CwlImportError(f"{where} has neither a baseCommand nor arguments, so there is nothing to run")
return tool
def _normalize_tool_outputs(saved: Mapping[str, object], where: str) -> dict[str, object]:
"""Normalize the outputs of one tool, which httk collects from the disk."""
outputs: dict[str, object] = {}
for entry in _entries(saved.get("outputs"), f"{where} outputs"):
name = _name(entry.get("id", ""))
location = f"{where}.{name}"
_reject_parameter_extras(entry, location)
declared = _normalize_type(entry.get("type"), location)
binding = entry.get("outputBinding")
output: dict[str, object] = {"type": declared}
if declared["type"] in {"stdout", "stderr"}:
outputs[name] = output
continue
if not isinstance(binding, Mapping):
raise UnsupportedCwlError(
f"unsupported CWL feature output without an outputBinding at {location}; "
"httk collects an output from a glob, and has no expression engine to compute one"
)
if binding.get("outputEval") is not None:
raise UnsupportedCwlError(
f"unsupported CWL feature outputEval at {location}; httk collects the files a glob matched, unchanged"
)
glob = binding.get("glob")
if glob is None:
raise UnsupportedCwlError(f"unsupported CWL feature outputBinding without a glob at {location}")
base = declared["items"] if declared["type"] == "array" else declared
if not isinstance(base, Mapping) or base.get("type") not in {"File", "Directory", "Any"}:
raise UnsupportedCwlError(
f"unsupported CWL feature output type {base!r} at {location}; "
"without an expression engine httk can collect File and Directory outputs only"
)
output["glob"] = [_check_expression(item, f"{location} glob") for item in _as_list(glob)]
if binding.get("loadContents"):
output["loadContents"] = True
outputs[name] = output
return outputs
def _normalize_workflow(saved: Mapping[str, object], where: str, context: CwlNotes) -> dict[str, object]:
"""Normalize the parts that only a ``Workflow`` has."""
declared = _entries(saved.get("steps"), f"{where} steps")
names = [_name(entry.get("id", "")) for entry in declared]
outputs: dict[str, object] = {}
for entry in _entries(saved.get("outputs"), f"{where} outputs"):
name = _name(entry.get("id", ""))
location = f"{where}.{name}"
_reject_parameter_extras(entry, location)
source = entry.get("outputSource")
if source is None:
raise UnsupportedCwlError(f"unsupported CWL feature workflow output without an outputSource at {location}")
outputs[name] = {
"type": _normalize_type(entry.get("type"), location),
"outputSource": [_source_name(item, names) for item in _as_list(source)],
"linkMerge": str(entry.get("linkMerge") or "merge_nested"),
}
steps: dict[str, object] = {}
for entry, name in zip(declared, names):
steps[name] = _normalize_step(entry, f"{where}.{name}", context, names)
if not steps:
raise CwlImportError(f"{where} has no steps, so there is nothing to run")
return {"outputs": outputs, "steps": steps}
def _normalize_step(
entry: Mapping[str, object],
where: str,
context: CwlNotes,
names: Sequence[str] = (),
) -> dict[str, object]:
"""Normalize one workflow step, its scatter and the process it runs."""
if entry.get("loop") is not None or str(entry.get("class", "")).endswith("LoopWorkflowStep"):
raise UnsupportedCwlError(f"unsupported CWL feature loop at {where}; httk has no CWL loop support")
method = entry.get("scatterMethod")
scatter = [_name(item) for item in _as_list(entry.get("scatter"))]
if method is not None and str(method) != "dotproduct":
raise UnsupportedCwlError(
f"unsupported CWL feature scatterMethod {method} at {where}; httk scatters one input, "
"or several of equal length with dotproduct"
)
if len(scatter) > 1 and method is None:
raise UnsupportedCwlError(
f"unsupported CWL feature scatter over {len(scatter)} inputs without scatterMethod at {where}; "
"httk supports dotproduct only"
)
when = entry.get("when")
if when is not None:
text = str(_check_expression(when, f"{where} when"))
if not _PLAIN_REFERENCE.fullmatch(text):
raise UnsupportedCwlError(
f"unsupported CWL feature conditional when at {where}: {text!r}; "
"httk evaluates one plain parameter reference, and nothing that has to be computed"
)
connections: dict[str, object] = {}
for item in _entries(entry.get("in"), f"{where} in"):
port = _name(item.get("id", ""))
location = f"{where}.{port}"
connection: dict[str, object] = {
"source": [_source_name(source, names) for source in _as_list(item.get("source"))]
}
if "default" in item and item["default"] is not None:
connection["default"] = _check_expression(item["default"], f"{location} default")
if item.get("valueFrom") is not None:
value = str(_check_expression(item["valueFrom"], f"{location} valueFrom"))
if "$(self" in value or "$(runtime" in value:
raise UnsupportedCwlError(
f"unsupported CWL feature valueFrom reference at {location}: {value!r}; "
"httk resolves $(inputs.name) references of a step's own inputs only"
)
connection["valueFrom"] = value
connection["linkMerge"] = str(item.get("linkMerge") or "merge_nested")
connections[port] = connection
run = entry.get("run")
if isinstance(run, str):
run = _saved_document(run, f"{where} run")
if not isinstance(run, Mapping):
raise CwlImportError(f"the run of {where} is neither a document nor a reference to one")
return {
"run": _normalize_process(run, f"{where} run", context),
"in": connections,
"out": [_name(item) for item in _as_list(entry.get("out"))],
"scatter": scatter,
"scatterMethod": "dotproduct" if method is None else str(method),
"when": None if when is None else str(when),
}
def _entries(value: object, where: str) -> list[Mapping[str, object]]:
"""Return one CWL array-or-map field as a list of objects carrying ids."""
if value is None:
return []
if isinstance(value, Mapping):
result: list[Mapping[str, object]] = []
for name, body in value.items():
if isinstance(body, Mapping):
result.append({"id": name, **body})
else:
result.append({"id": name, "type": body})
return result
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
entries = []
for item in value:
if isinstance(item, Mapping):
entries.append(item)
elif isinstance(item, str):
entries.append({"id": item})
else:
raise CwlImportError(f"one entry of {where} is neither an object nor a name")
return entries
raise CwlImportError(f"{where} is neither an array nor an object")
def _as_list(value: object) -> list[object]:
"""Return one CWL scalar-or-array member as a list."""
if value is None:
return []
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
return list(value)
return [value]
# ---------------------------------------------------------------------------
# cwl-utils, and the documents it reads
# ---------------------------------------------------------------------------
def _parser() -> Any:
"""Return :mod:`cwl_utils.parser`, or say exactly what to install."""
if find_spec("cwl_utils") is None:
raise CwlImportError(
"importing CWL needs the cwl-utils parser, which is an optional dependency of httk-workflow: "
f"install it with `{_CWL_EXTRA}`"
)
return import_module("cwl_utils.parser")
def _saved_document(uri: str, where: str) -> Mapping[str, object]:
"""Load one referenced CWL document and return it as saved plain JSON."""
parser = _parser()
try:
loaded = parser.load_document_by_uri(uri)
except Exception as exc:
raise CwlImportError(f"cannot parse the CWL document referenced at {where} ({uri}): {exc}") from exc
return _saved(loaded, uri)
def _saved(loaded: object, uri: str) -> Mapping[str, object]:
"""Return one parsed cwl-utils object as the plain JSON document it saves to."""
parser = _parser()
if isinstance(loaded, Sequence) and not isinstance(loaded, (str, bytes)):
# A ``$graph`` document: only a named entry point can be executed.
candidates = {str(getattr(item, "id", "")): item for item in loaded}
for identifier, item in candidates.items():
if identifier.endswith("#main"):
loaded = item
break
else:
raise UnsupportedCwlError(
f"unsupported CWL feature $graph without a #main entry point in {uri}; "
f"httk runs one process per document (found {', '.join(sorted(candidates)) or 'nothing'})"
)
saved = parser.save(loaded, relative_uris=False)
if not isinstance(saved, Mapping):
raise CwlImportError(f"cwl-utils did not save {uri} as one document")
return saved
def _upgraded(path: Path, scratch: Path) -> Path:
"""Return a v1.2 copy of the CWL tree around *path*, when one can be made.
*cwl-upgrader* rewrites one document at a time, and a workflow's ``run``
references point at its siblings, so the whole tree of ``.cwl`` files is
mirrored into *scratch* — a temporary directory, never the author's own — and
upgraded there, which keeps every relative reference resolving. Without the
package installed, or when the upgrade fails, the original is parsed as it
is: cwl-utils reads v1.0 and v1.1 too, and the subset this importer accepts
means the same thing in all three versions.
"""
if find_spec("cwlupgrader") is None:
return path
try:
upgrader = import_module("cwlupgrader.main")
root = path.parent.resolve()
for source in sorted(root.rglob("*.cwl")):
destination = scratch / source.relative_to(root)
destination.parent.mkdir(parents=True, exist_ok=True)
document = upgrader.load_cwl_document(str(source))
upgraded = upgrader.upgrade_document(document, str(destination.parent), "v1.2")
upgrader.write_cwl_document(upgraded, str(destination))
return scratch / path.resolve().relative_to(root)
except Exception as exc: # pragma: no cover - depends on the installed package
_LOGGER.debug("cannot upgrade %s to CWL v1.2, parsing it as it is: %s", path, exc)
return path
[docs]
def load_cwl_plan(workflow_path: str | os.PathLike[str]) -> tuple[dict[str, object], CwlNotes]:
"""Parse, check and normalize one CWL document into the plan httk executes."""
path = Path(workflow_path).expanduser()
if not path.is_file():
raise CwlImportError(f"the CWL document {path} does not exist")
parser = _parser()
version = _declared_version(path)
with TemporaryDirectory(prefix="httk-cwl-") as scratch:
source = path if version in {None, "v1.2"} else _upgraded(path, Path(scratch))
try:
loaded = parser.load_document_by_uri(str(source.resolve()))
except Exception as exc:
raise CwlImportError(f"cannot parse the CWL document {path}: {exc}") from exc
# Saving happens inside the scratch directory's life as well: a referenced
# document is loaded from beside the one that referenced it.
saved = _saved(loaded, str(path))
context = CwlNotes()
plan = _normalize_process(saved, path.name, context)
plan["cwlVersion"] = str(saved.get("cwlVersion") or version or "v1.2")
plan["source"] = path.name
return plan, context
def _declared_version(path: Path) -> str | None:
"""Return the ``cwlVersion`` one document declares, read as text."""
for line in path.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped.startswith("cwlVersion:"):
return stripped.partition(":")[2].strip().strip("\"'")
if stripped.startswith('"cwlVersion"'):
return stripped.partition(":")[2].strip().strip(" ,\"'")
return None
# ---------------------------------------------------------------------------
# The input object
# ---------------------------------------------------------------------------
def _yaml_load(text: str, path: Path) -> object:
"""Parse one YAML input object with the reader *cwl-utils* brings."""
parser = _parser()
try:
yaml = parser.yaml_no_ts()
return yaml.load(text)
except Exception as exc:
raise CwlImportError(f"cannot read the CWL input object {path} as YAML: {exc}") from exc
def _local_path(value: Mapping[str, object], base: Path, where: str) -> Path:
"""Return the local file or directory one File or Directory value names."""
location = value.get("path") or value.get("location")
if not isinstance(location, str) or not location:
raise CwlImportError(f"the {value.get('class')} input at {where} has neither a path nor a location")
if location.startswith("file://"):
location = unquote(urlparse(location).path)
elif "://" in location:
raise UnsupportedCwlError(
f"unsupported CWL feature remote location at {where}: {location}; httk stages local files only"
)
path = Path(location).expanduser()
if not path.is_absolute():
path = base / path
if not path.exists():
raise CwlImportError(f"the {value.get('class')} input at {where} does not exist: {path}")
return path
# ---------------------------------------------------------------------------
# The import itself
# ---------------------------------------------------------------------------
[docs]
def import_cwl(
workspace: Workspace,
workflow_path: str | os.PathLike[str],
inputs_path: str | os.PathLike[str],
*,
placement: str | PurePosixPath = DEFAULT_PLACEMENT,
tag: str | None = None,
name: str | None = None,
priority: int | None = None,
data_mode: Literal["none", "transactional"] = "none",
maximum_attempts: int | None = 3,
timeout: float | None = None,
) -> ImportedCwl:
"""Import one CWL document and its input object as one submitted job.
The document is parsed with *cwl-utils*, checked against the supported
subset, normalized into one self-contained JSON plan with every ``run:``
reference inlined, and staged in the payload together with the input object
and every local file it names. The job runs the packaged ``cwl_runner.py``
through the reserved installed form, so no runner file is written per
workflow.
Anything outside the subset raises :exc:`~httk.workflow.compat.cwl.UnsupportedCwlError` naming the
feature and where it was found; anything accepted with a caveat — a
``DockerRequirement``, an ignored hint — is reported in
:attr:`~httk.workflow.compat.cwl.ImportedCwl.warnings`.
"""
plan, context = load_cwl_plan(workflow_path)
values = load_cwl_inputs(inputs_path)
documents: dict[str, str] = {}
files: dict[str, str | os.PathLike[str]] = {}
staged = stage_cwl_inputs(values, Path(os.fspath(inputs_path)).expanduser().resolve().parent, documents, files)
documents[DOCUMENT_FILE] = json.dumps(plan, indent=1, sort_keys=True)
documents[INPUTS_FILE] = json.dumps(staged, indent=1, sort_keys=True)
inputs: dict[str, object] = {"cwl_document": DOCUMENT_FILE, "cwl_inputs": INPUTS_FILE}
if timeout is not None:
inputs["cwl_timeout"] = float(timeout)
job = submit_integration_job(
workspace,
runner_package=PACKAGE,
runner=RUNNER,
workflow=WORKFLOW,
initial_step=INITIAL_STEP,
name=name or f"cwl: {tag or Path(os.fspath(workflow_path)).name}",
inputs=inputs,
documents=documents,
files=files,
tag=tag,
placement=placement,
priority=priority,
data_mode=data_mode,
required_capabilities=tuple(sorted(context.capabilities)),
maximum_attempts_per_activation=maximum_attempts,
template="cwl",
)
for warning in context.warnings:
_LOGGER.warning("%s", warning)
return ImportedCwl(job=job, document=plan, inputs=staged, warnings=tuple(context.warnings))