docker-extract & cite-extract: support CWL v1.1 and v1.2

Co-authored-by: Michael R. Crusoe <[email protected]>

Author: manabuishii

Makefile

@@ -79,7 +79,7 @@ docs: FORCE
 
 ## clean                  : clean up all temporary / machine-generated files
 clean: FORCE
-	rm -f ${MODILE}/*.pyc tests/*.pyc
+	rm -f ${MODULE}/*.pyc tests/*.pyc
 	python setup.py clean --all || true
 	rm -Rf .coverage
 	rm -f diff-cover.html

create_cwl_from_objects.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 # SPDX-License-Identifier: Apache-2.0
-from ruamel import yaml
+import sys
+
+import ruamel.yaml
 
 from cwl_utils.parser import cwl_v1_2 as cwl
 
@@ -23,7 +25,8 @@ def main() -> None:
         stdin="$(inputs.file1.path)",
         stdout="output",
     )
-    print(yaml.main.round_trip_dump(cat_tool.save()))
+    yaml = ruamel.yaml.YAML()
+    yaml.dump(cat_tool.save(), sys.stdout)
 
 
 if __name__ == "__main__":

cwl_utils/cite_extract.py

@@ -2,11 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 import argparse
 import sys
-from typing import Iterator, List, Union, cast
+from typing import Iterator, cast
 
-import cwl_utils.parser.cwl_v1_0 as cwl
-
-ProcessType = Union[cwl.Workflow, cwl.CommandLineTool, cwl.ExpressionTool]
+import cwl_utils.parser as cwl
 
 
 def arg_parser() -> argparse.ArgumentParser:
@@ -21,50 +19,30 @@ def arg_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def parse_args(args: List[str]) -> argparse.Namespace:
-    """Parse the command line arguments."""
-    return arg_parser().parse_args(args)
-
-
 def run(args: argparse.Namespace) -> int:
     """Extract the software requirements."""
-    top = cwl.load_document(args.input)
-    traverse(top)
+    for req in traverse(cwl.load_document_by_uri(args.input)):
+        process_software_requirement(req)
     return 0
 
 
-def main() -> None:
+def main() -> int:
     """Console entry point."""
-    sys.exit(run(parse_args(sys.argv[1:])))
-
-
-def extract_software_packages(process: ProcessType) -> None:
-    """Print software packages found in the given process."""
-    for req in extract_software_reqs(process):
-        print(process.id)
-        process_software_requirement(req)
+    return run(arg_parser().parse_args(sys.argv[1:]))
 
 
 def extract_software_reqs(
-    process: ProcessType,
+    process: cwl.Process,
 ) -> Iterator[cwl.SoftwareRequirement]:
     """Return an iterator over any SoftwareRequirements found in the given process."""
     if process.requirements:
         for req in process.requirements:
-            if isinstance(req, cwl.SoftwareRequirement):
+            if isinstance(req, cwl.SoftwareRequirementTypes):
                 yield req
     if process.hints:
         for req in process.hints:
-            if isinstance(req, cwl.ProcessRequirement):
-                if isinstance(req, cwl.SoftwareRequirement):
-                    yield req
-            elif req["class"] == "SoftwareRequirement":
-                yield cwl.load_field(
-                    req,
-                    cwl.SoftwareRequirementLoader,
-                    process.id if process.id else "",
-                    process.loadingOptions,
-                )
+            if isinstance(req, cwl.SoftwareRequirementTypes):
+                yield req
 
 
 def process_software_requirement(req: cwl.SoftwareRequirement) -> None:
@@ -77,26 +55,26 @@ def process_software_requirement(req: cwl.SoftwareRequirement) -> None:
         )
 
 
-def traverse(process: ProcessType) -> None:
+def traverse(process: cwl.Process) -> Iterator[cwl.SoftwareRequirement]:
     """Extract the software packages for this process, and any steps."""
-    extract_software_packages(process)
-    if isinstance(process, cwl.Workflow):
-        traverse_workflow(process)
+    yield from extract_software_reqs(process)
+    if isinstance(process, cwl.WorkflowTypes):
+        yield from traverse_workflow(process)
 
 
-def get_process_from_step(step: cwl.WorkflowStep) -> ProcessType:
+def get_process_from_step(step: cwl.WorkflowStep) -> cwl.Process:
     """Return the process for this step, loading it if needed."""
     if isinstance(step.run, str):
-        return cast(ProcessType, cwl.load_document(step.run))
-    return cast(ProcessType, step.run)
+        return cast(cwl.Process, cwl.load_document_by_uri(step.run))
+    return cast(cwl.Process, step.run)
 
 
-def traverse_workflow(workflow: cwl.Workflow) -> None:
+def traverse_workflow(workflow: cwl.Workflow) -> Iterator[cwl.SoftwareRequirement]:
     """Iterate over the given workflow, extracting the software packages."""
     for step in workflow.steps:
-        extract_software_packages(step)
-        traverse(get_process_from_step(step))
+        yield from extract_software_reqs(step)
+        yield from traverse(get_process_from_step(step))
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

cwl_utils/docker_extract.py

@@ -3,18 +3,17 @@
 import argparse
 import os
 import sys
-from pathlib import Path
-from typing import Iterator, List, Union, cast
+from typing import Iterator, List, cast
 
-import cwl_utils.parser.cwl_v1_0 as cwl
+import ruamel.yaml
+
+import cwl_utils.parser as cwl
 from cwl_utils.image_puller import (
     DockerImagePuller,
     ImagePuller,
     SingularityImagePuller,
 )
 
-ProcessType = Union[cwl.Workflow, cwl.CommandLineTool, cwl.ExpressionTool]
-
 
 def arg_parser() -> argparse.ArgumentParser:
     """Argument parser."""
@@ -35,41 +34,51 @@ def arg_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--container-engine",
         dest="container_engine",
-        default="docker",
-        help="Specify which command to use to run OCI containers.",
+        help="Specify which command to use to run OCI containers. "
+        "Defaults to 'docker' (or 'singularity' if --singularity/-s is passed).",
     )
     return parser
 
 
-def parse_args(args: List[str]) -> argparse.Namespace:
-    """Parse the command line arguments."""
-    return arg_parser().parse_args(args)
-
-
-def run(args: argparse.Namespace) -> int:
-    """Extract the docker reqs and download them using Singularity or docker."""
+def run(args: argparse.Namespace) -> List[cwl.DockerRequirement]:
+    """Extract the docker reqs and download them using Singularity or Docker."""
     os.makedirs(args.dir, exist_ok=True)
 
-    top = cwl.load_document(args.input)
+    top = cwl.load_document_by_uri(args.input)
+    reqs: List[cwl.DockerRequirement] = []
 
     for req in traverse(top):
+        reqs.append(req)
         if not req.dockerPull:
-            print(f"Unable to save image from {req} due to lack of 'dockerPull'.")
+            print(
+                "Unable to save image from due to lack of 'dockerPull':",
+                file=sys.stderr,
+            )
+            yaml = ruamel.yaml.YAML()
+            yaml.dump(req.save(), sys.stderr)
             continue
         if args.singularity:
             image_puller: ImagePuller = SingularityImagePuller(
-                req.dockerPull, args.dir, "singularity"
+                req.dockerPull,
+                args.dir,
+                args.container_engine
+                if args.container_engine is not None
+                else "singularity",
             )
         else:
             image_puller = DockerImagePuller(
-                req.dockerPull, args.dir, args.container_engine
+                req.dockerPull,
+                args.dir,
+                args.container_engine
+                if args.container_engine is not None
+                else "docker",
             )
         image_puller.save_docker_image()
-    return 0
+    return reqs
 
 
 def extract_docker_requirements(
-    process: ProcessType,
+    process: cwl.Process,
 ) -> Iterator[cwl.DockerRequirement]:
     """Yield an iterator of the docker reqs, normalizing the pull request."""
     for req in extract_docker_reqs(process):
@@ -78,38 +87,30 @@ def extract_docker_requirements(
         yield req
 
 
-def extract_docker_reqs(process: ProcessType) -> Iterator[cwl.DockerRequirement]:
+def extract_docker_reqs(process: cwl.Process) -> Iterator[cwl.DockerRequirement]:
     """For the given process, extract the DockerRequirement(s)."""
     if process.requirements:
         for req in process.requirements:
-            if isinstance(req, cwl.DockerRequirement):
+            if isinstance(req, cwl.DockerRequirementTypes):
                 yield req
     if process.hints:
         for req in process.hints:
-            if isinstance(req, cwl.ProcessRequirement):
-                if isinstance(req, cwl.DockerRequirement):
-                    yield req
-            elif req["class"] == "DockerRequirement":
-                yield cwl.load_field(
-                    req,
-                    cwl.DockerRequirementLoader,
-                    Path.cwd().as_uri(),
-                    process.loadingOptions,
-                )
-
-
-def traverse(process: ProcessType) -> Iterator[cwl.DockerRequirement]:
+            if isinstance(req, cwl.DockerRequirementTypes):
+                yield req
+
+
+def traverse(process: cwl.Process) -> Iterator[cwl.DockerRequirement]:
     """Yield the iterator for the docker reqs, including an workflow steps."""
     yield from extract_docker_requirements(process)
-    if isinstance(process, cwl.Workflow):
+    if isinstance(process, cwl.WorkflowTypes):
         yield from traverse_workflow(process)
 
 
-def get_process_from_step(step: cwl.WorkflowStep) -> ProcessType:
+def get_process_from_step(step: cwl.WorkflowStep) -> cwl.Process:
     """Return the process for this step, loading it if necessary."""
     if isinstance(step.run, str):
-        return cast(ProcessType, cwl.load_document(step.run))
-    return cast(ProcessType, step.run)
+        return cast(cwl.Process, cwl.load_document_by_uri(step.run))
+    return cast(cwl.Process, step.run)
 
 
 def traverse_workflow(workflow: cwl.Workflow) -> Iterator[cwl.DockerRequirement]:
@@ -119,10 +120,11 @@ def traverse_workflow(workflow: cwl.Workflow) -> Iterator[cwl.DockerRequirement]
         yield from traverse(get_process_from_step(step))
 
 
-def main() -> None:
+def main() -> int:
     """Command line entry point."""
-    sys.exit(run(parse_args(sys.argv[1:])))
+    run(arg_parser().parse_args(sys.argv[1:]))
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

cwl_utils/image_puller.py

@@ -4,7 +4,8 @@
 import os
 import subprocess  # nosec
 from abc import ABC, abstractmethod
-from typing import List
+from pathlib import Path
+from typing import List, Union
 
 from .singularity import get_version as get_singularity_version
 from .singularity import is_version_2_6 as is_singularity_version_2_6
@@ -15,7 +16,7 @@
 
 
 class ImagePuller(ABC):
-    def __init__(self, req: str, save_directory: str, cmd: str) -> None:
+    def __init__(self, req: str, save_directory: Union[str, Path], cmd: str) -> None:
         """Create an ImagePuller."""
         self.req = req
         self.save_directory = save_directory
@@ -81,10 +82,6 @@ class SingularityImagePuller(ImagePuller):
     CHARS_TO_REPLACE = ["/"]
     NEW_CHAR = "_"
 
-    def __init__(self, req: str, save_directory: str, cmd: str) -> None:
-        """Create a Singularity-based software container image downloader."""
-        super().__init__(req, save_directory, cmd)
-
     def get_image_name(self) -> str:
         """Determine the file name appropriate to the installed version of Singularity."""
         image_name = self.req

cwl_utils/parser/__init__.py

@@ -85,6 +85,28 @@
 """Type union for a CWL v1.x DockerRequirement object."""
 Process = Union[Workflow, CommandLineTool, ExpressionTool, cwl_v1_2.Operation]
 """Type Union for a CWL v1.x Process object."""
+ProcessRequirement = Union[
+    cwl_v1_0.ProcessRequirement,
+    cwl_v1_1.ProcessRequirement,
+    cwl_v1_2.ProcessRequirement,
+]
+"""Type Union for a CWL v1.x ProcessRequirement object."""
+ProcessRequirementTypes = (
+    cwl_v1_0.ProcessRequirement,
+    cwl_v1_1.ProcessRequirement,
+    cwl_v1_2.ProcessRequirement,
+)
+SoftwareRequirement = Union[
+    cwl_v1_0.SoftwareRequirement,
+    cwl_v1_1.SoftwareRequirement,
+    cwl_v1_2.SoftwareRequirement,
+]
+SoftwareRequirementTypes = (
+    cwl_v1_0.SoftwareRequirement,
+    cwl_v1_1.SoftwareRequirement,
+    cwl_v1_2.SoftwareRequirement,
+)
+"""Type union for a CWL v1.x SoftwareRequirement object."""
 ArraySchema = Union[cwl_v1_0.ArraySchema, cwl_v1_1.ArraySchema, cwl_v1_2.ArraySchema]
 """Type Union for a CWL v1.x ArraySchema object."""
 EnumSchema = Union[cwl_v1_0.EnumSchema, cwl_v1_1.EnumSchema, cwl_v1_2.EnumSchema]
@@ -103,6 +125,7 @@
 """Type Union for a CWL v1.x Dirent object."""
 
 _Loader = Union[cwl_v1_0._Loader, cwl_v1_1._Loader, cwl_v1_2._Loader]
+"""Type union for a CWL v1.x _Loader."""
 
 
 def _get_id_from_graph(yaml: MutableMapping[str, Any], id_: Optional[str]) -> Any:

cwl_utils/parser/cwl_v1_0_utils.py

@@ -4,7 +4,6 @@
 from collections import namedtuple
 from typing import IO, Any, Dict, List, MutableSequence, Optional, Tuple, Union, cast
 
-from ruamel import yaml
 from schema_salad.exceptions import ValidationException
 from schema_salad.sourceline import SourceLine
 from schema_salad.utils import aslist, json_dumps
@@ -13,6 +12,7 @@
 import cwl_utils.parser.cwl_v1_0 as cwl
 import cwl_utils.parser.utils
 from cwl_utils.errors import WorkflowException
+from cwl_utils.utils import yaml_dumps
 
 CONTENT_LIMIT: int = 64 * 1024
 
@@ -297,7 +297,7 @@ def type_for_step_output(
     raise ValidationException(
         "param {} not found in {}.".format(
             sourcename,
-            yaml.main.round_trip_dump(cwl.save(step_run)),
+            yaml_dumps(cwl.save(step_run)),
         )
     )
 
@@ -428,9 +428,7 @@ def param_for_source_id(
     raise WorkflowException(
         "param {} not found in {}\n{}.".format(
             sourcename,
-            yaml.main.round_trip_dump(cwl.save(process)),
-            f" or\n {yaml.main.round_trip_dump(cwl.save(parent))}"
-            if parent is not None
-            else "",
+            yaml_dumps(cwl.save(process)),
+            f" or\n {yaml_dumps(cwl.save(parent))}" if parent is not None else "",
         )
     )