Store cross-version processes in an external context dict

src/cwl_utils/parser/__init__.py

@@ -193,8 +193,10 @@ class NoType(ABC):
     cwl_v1_2.DockerRequirement,
 )
 """Type union for a CWL v1.x DockerRequirement object."""
-Process: TypeAlias = Workflow | CommandLineTool | ExpressionTool | cwl_v1_2.Operation
+AbstractProcess: TypeAlias = cwl_v1_0.Process | cwl_v1_1.Process | cwl_v1_2.Process
 """Type Union for a CWL v1.x Process object."""
+Process: TypeAlias = Workflow | CommandLineTool | ExpressionTool | cwl_v1_2.Operation
+"""Type Union for a CWL v1.x Process implementations."""
 ProcessRequirement: TypeAlias = (
     cwl_v1_0.ProcessRequirement
     | cwl_v1_1.ProcessRequirement
@@ -449,7 +451,7 @@ def save(
 
 def is_process(v: Any) -> bool:
     """Test to see if the object is a CWL v1.x Python Process object."""
-    return isinstance(v, cwl_v1_0.Process | cwl_v1_1.Process | cwl_v1_2.Process)
+    return isinstance(v, AbstractProcess)
 
 
 def version_split(version: str) -> MutableSequence[int]:

src/cwl_utils/parser/cwl_v1_0_utils.py

@@ -265,10 +265,9 @@ def type_for_step_input(
     """Determine the type for the given step input."""
     if in_.valueFrom is not None:
         return "Any"
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.inputs:
-        for step_input in step_run.inputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []:
             if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]:
                 input_type = step_input.type_
                 if step.scatter is not None and in_.id in aslist(step.scatter):
@@ -282,10 +281,9 @@ def type_for_step_output(
     sourcename: str,
 ) -> Any:
     """Determine the type for the given step output."""
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.outputs:
-        for step_output in step_run.outputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for step_output in cast(cwl_utils.parser.Process, step_run).outputs or []:
             if (
                 step_output.id.split("#")[-1].split("/")[-1]
                 == sourcename.split("#")[-1].split("/")[-1]
@@ -311,11 +309,12 @@ def type_for_source(
     sourcenames: str | list[str],
     parent: cwl.Workflow | None = None,
     linkMerge: str | None = None,
+    loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None,
 ) -> Any:
     """Determine the type for the given sourcenames."""
     scatter_context: list[tuple[int, str] | None] = []
     params = cwl_utils.parser.utils.param_for_source_id(
-        process, sourcenames, parent, scatter_context
+        process, sourcenames, parent, scatter_context, loaded_steps
     )
     if not isinstance(params, MutableSequence):
         new_type = params.type_

src/cwl_utils/parser/cwl_v1_1_utils.py

@@ -347,10 +347,9 @@ def type_for_step_input(
     """Determine the type for the given step input."""
     if in_.valueFrom is not None:
         return "Any"
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.inputs:
-        for step_input in step_run.inputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []:
             if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]:
                 input_type = step_input.type_
                 if step.scatter is not None and in_.id in aslist(step.scatter):
@@ -364,10 +363,9 @@ def type_for_step_output(
     sourcename: str,
 ) -> Any:
     """Determine the type for the given step output."""
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.outputs:
-        for output in step_run.outputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for output in cast(cwl_utils.parser.Process, step_run).outputs or []:
             if (
                 output.id.split("#")[-1].split("/")[-1]
                 == sourcename.split("#")[-1].split("/")[-1]
@@ -393,11 +391,12 @@ def type_for_source(
     sourcenames: str | list[str],
     parent: cwl.Workflow | None = None,
     linkMerge: str | None = None,
+    loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None,
 ) -> Any:
     """Determine the type for the given sourcenames."""
     scatter_context: list[tuple[int, str] | None] = []
     params = cwl_utils.parser.utils.param_for_source_id(
-        process, sourcenames, parent, scatter_context
+        process, sourcenames, parent, scatter_context, loaded_steps
     )
     if not isinstance(params, MutableSequence):
         new_type = params.type_

src/cwl_utils/parser/cwl_v1_2_utils.py

@@ -375,10 +375,9 @@ def type_for_step_input(
     """Determine the type for the given step input."""
     if in_.valueFrom is not None:
         return "Any"
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.inputs:
-        for step_input in step_run.inputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []:
             if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]:
                 input_type = step_input.type_
                 if step.scatter is not None and in_.id in aslist(step.scatter):
@@ -392,10 +391,9 @@ def type_for_step_output(
     sourcename: str,
 ) -> Any:
     """Determine the type for the given step output."""
-    step_run = cwl_utils.parser.utils.load_step(step)
-    cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
-    if step_run and step_run.outputs:
-        for output in step_run.outputs:
+    if step_run := cwl_utils.parser.utils.load_step(step):
+        cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
+        for output in cast(cwl_utils.parser.Process, step_run).outputs or []:
             if (
                 output.id.split("#")[-1].split("/")[-1]
                 == sourcename.split("#")[-1].split("/")[-1]
@@ -422,11 +420,12 @@ def type_for_source(
     parent: cwl.Workflow | None = None,
     linkMerge: str | None = None,
     pickValue: str | None = None,
+    loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None,
 ) -> Any:
     """Determine the type for the given sourcenames."""
     scatter_context: list[tuple[int, str] | None] = []
     params = cwl_utils.parser.utils.param_for_source_id(
-        process, sourcenames, parent, scatter_context
+        process, sourcenames, parent, scatter_context, loaded_steps
     )
     if not isinstance(params, MutableSequence):
         new_type = params.type_

src/cwl_utils/parser/utils.py

@@ -32,9 +32,10 @@
     CommandOutputParameter,
     WorkflowInputParameter,
     load_document_by_uri,
+    AbstractProcess,
 )
 from cwl_utils.errors import WorkflowException
-from cwl_utils.utils import yaml_dumps
+from cwl_utils.utils import yaml_dumps, get_step_uri
 
 _logger = logging.getLogger("cwl_utils")
 
@@ -144,7 +145,7 @@ def check_types(
             raise ValidationException(f"Invalid value {linkMerge} for linkMerge field.")
 
 
-def convert_stdstreams_to_files(process: Process) -> None:
+def convert_stdstreams_to_files(process: AbstractProcess) -> None:
     """Convert stdin, stdout and stderr type shortcuts to files."""
     match process:
         case cwl_v1_0.CommandLineTool():
@@ -230,18 +231,28 @@ def load_inputfile_by_yaml(
 
 
 def load_step(
-    step: WorkflowStep,
-) -> Process:
+    step: WorkflowStep, loaded_steps: dict[str, AbstractProcess] | None = None
+) -> AbstractProcess:
     if isinstance(step.run, str):
-        step_run = load_document_by_uri(
-            path=step.loadingOptions.fetcher.urljoin(
-                base_url=cast(str, step.loadingOptions.fileuri),
-                url=step.run,
-            ),
-            loadingOptions=step.loadingOptions,
-        )
-        return cast(Process, step_run)
-    return cast(Process, copy.deepcopy(step.run))
+        uri = get_step_uri(step)
+        if loaded_steps is not None and uri in loaded_steps:
+            return loaded_steps[uri]
+        else:
+            step_run = cast(
+                AbstractProcess,
+                load_document_by_uri(
+                    path=uri,
+                    loadingOptions=step.loadingOptions,
+                ),
+            )
+            if loaded_steps is not None:
+                loaded_steps[uri] = step_run
+            return step_run
+    else:
+        step_run = copy.deepcopy(step.run)
+        if not isinstance(step_run, cwl_utils.parser.Process):
+            raise Exception(f"Unsupported process type: {step_run.__class__.__name__}")
+        return step_run
 
 
 def merge_flatten_type(src: Any) -> Any:
@@ -258,6 +269,7 @@ def param_for_source_id(
     sourcenames: str | list[str],
     parent: Workflow | None = None,
     scatter_context: list[tuple[int, str] | None] | None = None,
+    loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None,
 ) -> (
     CommandInputParameter
     | CommandOutputParameter
@@ -295,7 +307,10 @@ def param_for_source_id(
                         == step.id.split("#")[-1]
                         and step.out
                     ):
-                        step_run = cwl_utils.parser.utils.load_step(step)
+                        step_run = cast(
+                            Process,
+                            cwl_utils.parser.utils.load_step(step, loaded_steps),
+                        )
                         cwl_utils.parser.utils.convert_stdstreams_to_files(step_run)
                         for outp in step.out:
                             outp_id = outp if isinstance(outp, str) else outp.id
@@ -527,6 +542,7 @@ def type_for_source(
     parent: Workflow | None = None,
     linkMerge: str | None = None,
     pickValue: str | None = None,
+    loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None,
 ) -> Any:
     """Determine the type for the given sourcenames."""
     match process.cwlVersion or cwlVersion:
@@ -541,6 +557,7 @@ def type_for_source(
                 sourcenames,
                 cast(cwl_v1_0.Workflow | None, parent),
                 linkMerge,
+                loaded_steps,
             )
         case "v1.1":
             return cwl_v1_1_utils.type_for_source(
@@ -553,6 +570,7 @@ def type_for_source(
                 sourcenames,
                 cast(cwl_v1_1.Workflow | None, parent),
                 linkMerge,
+                loaded_steps,
             )
         case "v1.2":
             return cwl_v1_2_utils.type_for_source(
@@ -566,6 +584,7 @@ def type_for_source(
                 cast(cwl_v1_2.Workflow | None, parent),
                 linkMerge,
                 pickValue,
+                loaded_steps,
             )
         case _ as cwlVersion:
             raise ValidationException(

src/cwl_utils/utils.py

@@ -11,7 +11,7 @@
 from copy import deepcopy
 from importlib.resources import files
 from io import StringIO
-from typing import Any
+from typing import Any, cast
 from urllib.parse import urlparse
 
 from ruamel.yaml.main import YAML
@@ -22,7 +22,7 @@
 from cwl_utils.loghandler import _logger
 
 # Type hinting
-from cwl_utils.parser import cwl_v1_0, cwl_v1_1, cwl_v1_2
+from cwl_utils.parser import cwl_v1_0, cwl_v1_1, cwl_v1_2, WorkflowStep
 
 # Load as 1.2 files
 from cwl_utils.parser.cwl_v1_2 import InputArraySchema as InputArraySchemaV1_2
@@ -454,6 +454,17 @@ def is_local_uri(uri: str) -> bool:
     return False
 
 
+def get_step_uri(step: WorkflowStep) -> str:
+    if not isinstance(step.run, str):
+        raise Exception(
+            f"Impossible to retrieve URI for step {step.id}: it embeds a process"
+        )
+    return step.loadingOptions.fetcher.urljoin(
+        base_url=cast(str, step.loadingOptions.fileuri),
+        url=step.run,
+    )
+
+
 def get_value_from_uri(uri: str) -> str:
     """
     Given a URI, return the value after #.