2025-12 Codegen refresh

cwl_utils/cwl_v1_0_expression_refactor.py

@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -445,7 +445,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -1815,9 +1815,9 @@ def traverse_step(
                 if not target:
                     raise WorkflowException("target not found")
                 input_source_id = None
-                source_type: None | (list[cwl.InputParameter] | cwl.InputParameter) = (
-                    None
-                )
+                source_type: (
+                    None | MutableSequence[cwl.InputParameter] | cwl.InputParameter
+                ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
                         input_source_id = []
@@ -1915,7 +1915,7 @@ def workflow_step_to_InputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     if not p.type_:
                         raise WorkflowException(
@@ -1946,7 +1946,7 @@ def replace_step_valueFrom_expr_with_etool(
     original_step_ins: list[cwl.WorkflowStepInput],
     source: str | list[str] | None,
     replace_etool: bool,
-    source_type: cwl.InputParameter | list[cwl.InputParameter] | None = None,
+    source_type: cwl.InputParameter | MutableSequence[cwl.InputParameter] | None = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""
     if not step_inp.id:

cwl_utils/cwl_v1_1_expression_refactor.py

@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -443,7 +443,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -1818,7 +1818,11 @@ def traverse_step(
                     raise WorkflowException("target not found")
                 input_source_id = None
                 source_type: None | (
-                    list[cwl.WorkflowInputParameter] | cwl.WorkflowInputParameter
+                    MutableSequence[
+                        cwl.CommandInputParameter | cwl.WorkflowInputParameter
+                    ]
+                    | cwl.CommandInputParameter
+                    | cwl.WorkflowInputParameter
                 ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
@@ -1906,7 +1910,7 @@ def traverse_step(
 
 def workflow_step_to_WorkflowInputParameters(
     step_ins: list[cwl.WorkflowStepInput], parent: cwl.Workflow, except_in_id: str
-) -> list[cwl.WorkflowInputParameter]:
+) -> MutableSequence[cwl.CommandInputParameter | cwl.WorkflowInputParameter]:
     """Create WorkflowInputParameters to match the given WorkflowStep inputs."""
     params = []
     for inp in step_ins:
@@ -1917,7 +1921,7 @@ def workflow_step_to_WorkflowInputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     p.id = inp_id
                     p.type_ = clean_type_ids(p.type_)
@@ -1941,7 +1945,9 @@ def replace_step_valueFrom_expr_with_etool(
     source: str | list[str] | None,
     replace_etool: bool,
     source_type: None | (
-        cwl.WorkflowInputParameter | list[cwl.WorkflowInputParameter]
+        cwl.CommandInputParameter
+        | cwl.WorkflowInputParameter
+        | MutableSequence[cwl.CommandInputParameter | cwl.WorkflowInputParameter]
     ) = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""

cwl_utils/cwl_v1_2_expression_refactor.py

@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import Mapping, MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -442,7 +442,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -725,7 +725,9 @@ def process_workflow_inputs_and_outputs(
                         source_type_items.append("null")
                 elif source_type_items != "null":
                     source_type_items = ["null", source_type_items]
-                source_type = cwl.CommandInputParameter(type_=source_type_items)
+                source_type = cwl.CommandInputParameter(
+                    id=None, type_=source_type_items
+                )
                 replace_expr_with_etool(
                     expression,
                     etool_id,
@@ -1919,7 +1921,11 @@ def traverse_step(
                     raise WorkflowException("target not found")
                 input_source_id = None
                 source_type: None | (
-                    list[cwl.WorkflowInputParameter] | cwl.WorkflowInputParameter
+                    MutableSequence[
+                        cwl.CommandInputParameter | cwl.WorkflowInputParameter
+                    ]
+                    | cwl.CommandInputParameter
+                    | cwl.WorkflowInputParameter
                 ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
@@ -2015,7 +2021,7 @@ def traverse_step(
 
 def workflow_step_to_WorkflowInputParameters(
     step_ins: list[cwl.WorkflowStepInput], parent: cwl.Workflow, except_in_id: str
-) -> list[cwl.WorkflowInputParameter]:
+) -> MutableSequence[cwl.CommandInputParameter | cwl.WorkflowInputParameter]:
     """Create WorkflowInputParameters to match the given WorkflowStep inputs."""
     params = []
     for inp in step_ins:
@@ -2026,7 +2032,7 @@ def workflow_step_to_WorkflowInputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     p.id = inp_id
                     p.type_ = clean_type_ids(p.type_)
@@ -2050,7 +2056,9 @@ def replace_step_valueFrom_expr_with_etool(
     source: str | list[str] | None,
     replace_etool: bool,
     source_type: None | (
-        cwl.WorkflowInputParameter | list[cwl.WorkflowInputParameter]
+        cwl.CommandInputParameter
+        | cwl.WorkflowInputParameter
+        | MutableSequence[cwl.CommandInputParameter | cwl.WorkflowInputParameter]
     ) = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""

cwl_utils/expression.py

@@ -6,7 +6,7 @@
 import json
 from collections.abc import Awaitable, MutableMapping
 from enum import Enum
-from typing import Any, Union, cast
+from typing import Any, cast
 
 from schema_salad.utils import json_dumps
 
@@ -293,7 +293,7 @@ def do_eval(
 
     :param timeout: The maximum number of seconds to wait while executing.
     """
-    runtime = cast(MutableMapping[str, Union[int, str, None]], copy.deepcopy(resources))
+    runtime = cast(MutableMapping[str, int | str | None], copy.deepcopy(resources))
     runtime["tmpdir"] = tmpdir or None
     runtime["outdir"] = outdir or None
 

cwl_utils/expression_refactor.py

@@ -8,7 +8,7 @@
 import sys
 from collections.abc import Callable, MutableMapping, MutableSequence
 from pathlib import Path
-from typing import Any, Optional, Protocol, Union
+from typing import Any, Protocol
 
 from ruamel.yaml.main import YAML
 from ruamel.yaml.scalarstring import walk_tree
@@ -28,9 +28,9 @@
 _logger.setLevel(logging.INFO)
 _cwlutilslogger.setLevel(100)
 
-save_type = Optional[
-    Union[MutableMapping[str, Any], MutableSequence[Any], int, float, bool, str]
-]
+save_type = (
+    MutableMapping[str, Any] | MutableSequence[Any] | int | float | bool | str | None
+)
 
 
 class saveCWL(Protocol):

cwl_utils/inputs_schema_gen.py

@@ -12,7 +12,7 @@
 from copy import deepcopy
 from importlib.resources import files
 from pathlib import Path
-from typing import Any, TypeGuard, Union
+from typing import Any, TypeGuard
 from urllib.parse import urlparse
 
 import requests
@@ -61,9 +61,9 @@
 }
 
 # Some type hinting
-InputType = Union[
-    InputArraySchema, InputEnumSchema, InputRecordSchema, str, File, Directory
-]
+InputType = (
+    InputArraySchema | InputEnumSchema | InputRecordSchema | str | File | Directory
+)
 
 
 # Don't need type checking at runtime

cwl_utils/parser/__init__.py

@@ -294,24 +294,49 @@ def load_document_by_uri(
             loadingOptions = cwl_v1_0.LoadingOptions(
                 fileuri=real_uri, baseuri=base_uri, copyfrom=loadingOptions
             )
+            return load_document_by_string(
+                loadingOptions.fetcher.fetch_text(real_uri),
+                real_uri,
+                loadingOptions,
+                id_,
+                load_all,
+            )
         case cwl_v1_1.LoadingOptions():
             loadingOptions = cwl_v1_1.LoadingOptions(
                 fileuri=real_uri, baseuri=base_uri, copyfrom=loadingOptions
             )
+            return load_document_by_string(
+                loadingOptions.fetcher.fetch_text(real_uri),
+                real_uri,
+                loadingOptions,
+                id_,
+                load_all,
+            )
         case cwl_v1_2.LoadingOptions():
             loadingOptions = cwl_v1_2.LoadingOptions(
                 fileuri=real_uri, baseuri=base_uri, copyfrom=loadingOptions
             )
+            return load_document_by_string(
+                loadingOptions.fetcher.fetch_text(real_uri),
+                real_uri,
+                loadingOptions,
+                id_,
+                load_all,
+            )
         case None:
             loadingOptions = cwl_v1_2.LoadingOptions(fileuri=real_uri, baseuri=base_uri)
+            return load_document_by_string(
+                loadingOptions.fetcher.fetch_text(real_uri),
+                real_uri,
+                None,
+                id_,
+                load_all,
+            )
         case _:
             raise ValidationException(
                 f"Unsupported loadingOptions type: {type(loadingOptions)}"
             )
 
-    doc = loadingOptions.fetcher.fetch_text(real_uri)
-    return load_document_by_string(doc, real_uri, loadingOptions, id_, load_all)
-
 
 def load_document(
     doc: Any,