use the new TypeDict types from cwl-utils 0.41 more (#2192)

Co-authored-by: Michael R. Crusoe <michael.crusoe@gmail.com>

Author: GlassOfWhiskey

cwltool/builder.py

@@ -94,7 +94,7 @@ class Builder(HasReqsHints):
     def __init__(
         self,
         job: CWLObjectType,
-        files: list[CWLFileType | CWLDirectoryType],
+        files: MutableSequence[CWLFileType | CWLDirectoryType],
         bindings: list[CWLObjectType],
         schemaDefs: MutableMapping[str, CWLObjectType],
         names: Names,
@@ -458,8 +458,8 @@ def _capture_files(f: CWLFileType) -> CWLFileType:
                             if not found:
 
                                 def addsf(
-                                    files: MutableSequence[CWLObjectType],
-                                    newsf: CWLObjectType,
+                                    files: MutableSequence[CWLFileType | CWLDirectoryType],
+                                    newsf: CWLFileType | CWLDirectoryType,
                                 ) -> None:
                                     for f in files:
                                         if f["location"] == newsf["location"]:
@@ -469,23 +469,19 @@ def addsf(
 
                                 if isinstance(sfname, MutableMapping):
                                     addsf(
-                                        cast(
-                                            MutableSequence[CWLObjectType],
-                                            fdatum["secondaryFiles"],
-                                        ),
-                                        sfname,
+                                        fdatum["secondaryFiles"],
+                                        cast(CWLFileType | CWLDirectoryType, sfname),
                                     )
                                 elif discover_secondaryFiles and self.fs_access.exists(sf_location):
                                     addsf(
-                                        cast(
-                                            MutableSequence[CWLObjectType],
-                                            fdatum["secondaryFiles"],
+                                        fdatum["secondaryFiles"],
+                                        CWLFileType(
+                                            **{
+                                                "location": sf_location,
+                                                "basename": sfname,
+                                                "class": "File",
+                                            }
                                         ),
-                                        {
-                                            "location": sf_location,
-                                            "basename": sfname,
-                                            "class": "File",
-                                        },
                                     )
                                 elif sf_required:
                                     raise SourceLine(

cwltool/checker.py

@@ -3,15 +3,15 @@
 from collections.abc import Iterator, MutableMapping, MutableSequence, Sized
 from typing import Any, Literal, NamedTuple, Optional, Union, cast
 
-from cwl_utils.types import CWLObjectType, CWLOutputType
+from cwl_utils.types import CWLObjectType, CWLOutputType, SinkType
 from schema_salad.exceptions import ValidationException
 from schema_salad.sourceline import SourceLine, bullets, strip_dup_lineno
 from schema_salad.utils import json_dumps
 
 from .errors import WorkflowException
 from .loghandler import _logger
 from .process import shortname
-from .utils import SinkType, aslist
+from .utils import aslist
 
 
 def _get_type(tp: Any) -> Any:
@@ -22,8 +22,8 @@ def _get_type(tp: Any) -> Any:
 
 
 def check_types(
-    srctype: SinkType,
-    sinktype: SinkType,
+    srctype: SinkType | None,
+    sinktype: SinkType | None,
     linkMerge: str | None,
     pickValue: str | None,
     valueFrom: str | None,
@@ -83,18 +83,20 @@ def check_types(
             raise WorkflowException(f"Unrecognized linkMerge enum {linkMerge!r}")
 
 
-def merge_flatten_type(src: SinkType) -> CWLOutputType:
+def merge_flatten_type(src: SinkType | None) -> CWLOutputType | None:
     """Return the merge flattened type of the source type."""
     match src:
         case MutableSequence():
-            return [merge_flatten_type(cast(SinkType, t)) for t in src]
+            return [merge_flatten_type(t) for t in src]
         case {"type": "array"}:
             return src
         case _:
             return {"items": src, "type": "array"}
 
 
-def can_assign_src_to_sink(src: SinkType, sink: SinkType | None, strict: bool = False) -> bool:
+def can_assign_src_to_sink(
+    src: SinkType | None, sink: SinkType | None, strict: bool = False
+) -> bool:
     """
     Check for identical type specifications, ignoring extra keys like inputBinding.
 
@@ -112,8 +114,8 @@ def can_assign_src_to_sink(src: SinkType, sink: SinkType | None, strict: bool =
             return False
         if src["type"] == "array" and sink["type"] == "array":
             return can_assign_src_to_sink(
-                cast(SinkType, src["items"]),
-                cast(SinkType, sink["items"]),
+                cast(MutableSequence[CWLOutputType | None], src["items"]),
+                cast(MutableSequence[CWLOutputType | None], sink["items"]),
                 strict,
             )
         if src["type"] == "record" and sink["type"] == "record":
@@ -128,15 +130,15 @@ def can_assign_src_to_sink(src: SinkType, sink: SinkType | None, strict: bool =
                     if strict:
                         return False
             return True
-        return can_assign_src_to_sink(cast(SinkType, src["type"]), sink["type"], strict)
+        return can_assign_src_to_sink(src["type"], sink["type"], strict)
     if isinstance(src, MutableSequence):
         if strict:
             for this_src in src:
-                if not can_assign_src_to_sink(cast(SinkType, this_src), sink):
+                if not can_assign_src_to_sink(this_src, sink):
                     return False
             return True
         for this_src in src:
-            if this_src != "null" and can_assign_src_to_sink(cast(SinkType, this_src), sink):
+            if this_src != "null" and can_assign_src_to_sink(this_src, sink):
                 return True
         return False
     if isinstance(sink, MutableSequence):