Typed CWL parsers

Author: GlassOfWhiskey

cwl_utils/cite_extract.py

@@ -33,7 +33,7 @@ def main() -> int:
 
 
 def extract_software_reqs(
-    process: cwl.Process,
+    process: cwl.Process | cwl.WorkflowStep,
 ) -> Iterator[cwl.SoftwareRequirement]:
     """Return an iterator over any SoftwareRequirements found in the given process."""
     if process.requirements:

cwl_utils/cwl_v1_0_expression_refactor.py

@@ -1465,6 +1465,7 @@ def traverse_CommandLineTool(
                         inp.linkMerge = None
                     for index, out in enumerate(new_clt_step.out):
                         new_clt_step.out[index] = out.split("/")[-1]
+
                     for tool_inp in new_clt_step.run.inputs:
                         tool_inp.id = tool_inp.id.split("#")[-1]
                     for tool_out in new_clt_step.run.outputs:

cwl_utils/docker_extract.py

@@ -102,7 +102,9 @@ def extract_docker_requirements(
         yield req
 
 
-def extract_docker_reqs(process: cwl.Process) -> Iterator[cwl.DockerRequirement]:
+def extract_docker_reqs(
+    process: cwl.Process | cwl.WorkflowStep,
+) -> Iterator[cwl.DockerRequirement]:
     """For the given process, extract the DockerRequirement(s)."""
     if process.requirements:
         for req in process.requirements:

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
+from typing import Any, TypeGuard, cast
 from urllib.parse import urlparse
 
 import requests
@@ -26,6 +26,7 @@
     InputEnumSchema,
     InputRecordSchema,
     InputRecordSchemaTypes,
+    SchemaDefRequirement,
     Workflow,
     WorkflowInputParameter,
     cwl_v1_0,
@@ -402,12 +403,15 @@ def get_complex_schema_values(idx_iter: str) -> InputRecordSchema:
 
     if cwl_obj.requirements is not None:
         with suppress(StopIteration):
-            schema_def_requirement = next(
-                filter(
-                    lambda requirement_iter: requirement_iter.class_
-                    == "SchemaDefRequirement",
-                    cwl_obj.requirements,
-                )
+            schema_def_requirement = cast(
+                SchemaDefRequirement,
+                next(
+                    filter(
+                        lambda requirement_iter: requirement_iter.class_
+                        == "SchemaDefRequirement",
+                        cwl_obj.requirements,
+                    )
+                ),
             )
 
             workflow_schema_definitions_list.extend(

cwl_utils/parser/__init__.py

@@ -238,6 +238,12 @@ class NoType(ABC):
     | cwl_v1_2.WorkflowStepInput
 )
 """Type Union for a CWL v1.x LoadContents object."""
+SchemaDefRequirement: TypeAlias = (
+    cwl_v1_0.SchemaDefRequirement
+    | cwl_v1_1.SchemaDefRequirement
+    | cwl_v1_2.SchemaDefRequirement
+)
+"""Type Union for a CWL v1.x SchemaDefRequirement object."""
 _Loader: TypeAlias = cwl_v1_0._Loader | cwl_v1_1._Loader | cwl_v1_2._Loader
 """Type union for a CWL v1.x _Loader."""