Experiment with TypedDict in CWL types

Makefile

@@ -25,7 +25,7 @@ MODULE=cwltool
 # `SHELL=bash` doesn't work for some, so don't use BASH-isms like
 # `[[` conditional expressions.
 PYSOURCES=$(wildcard ${MODULE}/**.py cwltool/cwlprov/*.py tests/*.py tests/cwl-conformance/*.py) setup.py
-DEVPKGS=diff_cover pylint pep257 pydocstyle 'tox<4' tox-pyenv auto-walrus \
+DEVPKGS=diff_cover pylint pep257 pydocstyle 'tox>4' auto-walrus \
 	isort wheel autoflake pyupgrade bandit -rlint-requirements.txt\
 	-rtest-requirements.txt -rmypy-requirements.txt -rdocs/requirements.txt
 DEBDEVPKGS=pep8 python-autopep8 pylint python-coverage pydocstyle sloccount \

cwltool/builder.py

@@ -9,6 +9,7 @@
 
 from cwl_utils import expression
 from cwl_utils.file_formats import check_format
+from cwl_utils.types import CWLDirectoryType, CWLFileType, CWLObjectType, CWLOutputType
 from mypy_extensions import mypyc_attr
 from rdflib import Graph
 from ruamel.yaml.comments import CommentedMap
@@ -26,8 +27,6 @@
 from .stdfsaccess import StdFsAccess
 from .utils import (
     CONTENT_LIMIT,
-    CWLObjectType,
-    CWLOutputType,
     HasReqsHints,
     LoadListingType,
     aslist,
@@ -95,7 +94,7 @@ class Builder(HasReqsHints):
     def __init__(
         self,
         job: CWLObjectType,
-        files: list[CWLObjectType],
+        files: MutableSequence[CWLFileType | CWLDirectoryType],
         bindings: list[CWLObjectType],
         schemaDefs: MutableMapping[str, CWLObjectType],
         names: Names,
@@ -166,7 +165,7 @@ def build_job_script(self, commands: list[str]) -> str | None:
             return self.job_script_provider.build_job_script(self, commands)
         return None
 
-    def _capture_files(self, f: CWLObjectType) -> CWLObjectType:
+    def _capture_files(self, f: CWLFileType | CWLDirectoryType) -> CWLFileType | CWLDirectoryType:
         self.files.append(f)
         return f
 
@@ -356,13 +355,13 @@ def bind_input(
                     )
                 binding = {}
 
-            def _capture_files(f: CWLObjectType) -> CWLObjectType:
+            def _capture_files(f: CWLFileType | CWLDirectoryType) -> CWLFileType | CWLDirectoryType:
                 self.files.append(f)
                 return f
 
             if schema["type"] == "org.w3id.cwl.cwl.File":
-                datum = cast(CWLObjectType, datum)
-                self.files.append(datum)
+                file_datum = cast(CWLFileType, datum)
+                self.files.append(file_datum)
 
                 loadContents_sourceline: (
                     None | MutableMapping[str, str | list[int]] | CWLObjectType
@@ -380,14 +379,16 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
                         debug,
                     ):
                         try:
-                            with self.fs_access.open(cast(str, datum["location"]), "rb") as f2:
-                                datum["contents"] = content_limit_respected_read(f2)
+                            with self.fs_access.open(file_datum["location"], "rb") as f2:
+                                file_datum["contents"] = content_limit_respected_read(f2)
                         except Exception as e:
-                            raise Exception("Reading {}\n{}".format(datum["location"], e)) from e
+                            raise Exception(
+                                "Reading {}\n{}".format(file_datum["location"], e)
+                            ) from e
 
                 if "secondaryFiles" in schema:
-                    if "secondaryFiles" not in datum:
-                        datum["secondaryFiles"] = []
+                    if "secondaryFiles" not in file_datum:
+                        file_datum["secondaryFiles"] = []
                         sf_schema = aslist(schema["secondaryFiles"])
                     elif not discover_secondaryFiles:
                         sf_schema = []  # trust the inputs
@@ -396,7 +397,7 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
 
                     for num, sf_entry in enumerate(sf_schema):
                         if "required" in sf_entry and sf_entry["required"] is not None:
-                            required_result = self.do_eval(sf_entry["required"], context=datum)
+                            required_result = self.do_eval(sf_entry["required"], context=file_datum)
                             if not (isinstance(required_result, bool) or required_result is None):
                                 if sf_schema == schema["secondaryFiles"]:
                                     sf_item: Any = sf_schema[num]
@@ -418,15 +419,15 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
                         if "$(" in sf_entry["pattern"] or "${" in sf_entry["pattern"]:
                             sfpath = self.do_eval(sf_entry["pattern"], context=datum)
                         else:
-                            sfpath = substitute(cast(str, datum["basename"]), sf_entry["pattern"])
+                            sfpath = substitute(file_datum["basename"], sf_entry["pattern"])
 
                         for sfname in aslist(sfpath):
                             if not sfname:
                                 continue
                             found = False
 
                             if isinstance(sfname, str):
-                                d_location = cast(str, datum["location"])
+                                d_location = file_datum["location"]
                                 if "/" in d_location:
                                     sf_location = (
                                         d_location[0 : d_location.rindex("/") + 1] + sfname
@@ -435,6 +436,7 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
                                     sf_location = d_location + sfname
                                 sfbasename = sfname
                             elif isinstance(sfname, MutableMapping):
+                                sfname = cast(CWLFileType | CWLDirectoryType, sfname)
                                 sf_location = sfname["location"]
                                 sfbasename = sfname["basename"]
                             else:
@@ -447,10 +449,7 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
                                     f"{type(sfname)!r} from {sf_entry['pattern']!r}."
                                 )
 
-                            for d in cast(
-                                MutableSequence[MutableMapping[str, str]],
-                                datum["secondaryFiles"],
-                            ):
+                            for d in file_datum["secondaryFiles"]:
                                 if not d.get("basename"):
                                     d["basename"] = d["location"][d["location"].rindex("/") + 1 :]
                                 if d["basename"] == sfbasename:
@@ -459,8 +458,8 @@ def _capture_files(f: CWLObjectType) -> CWLObjectType:
                             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"]:
@@ -470,23 +469,19 @@ def addsf(
 
                                 if isinstance(sfname, MutableMapping):
                                     addsf(
-                                        cast(
-                                            MutableSequence[CWLObjectType],
-                                            datum["secondaryFiles"],
-                                        ),
-                                        sfname,
+                                        file_datum["secondaryFiles"],
+                                        cast(CWLFileType | CWLDirectoryType, sfname),
                                     )
                                 elif discover_secondaryFiles and self.fs_access.exists(sf_location):
                                     addsf(
-                                        cast(
-                                            MutableSequence[CWLObjectType],
-                                            datum["secondaryFiles"],
+                                        file_datum["secondaryFiles"],
+                                        CWLFileType(
+                                            **{
+                                                "location": sf_location,
+                                                "basename": sfname,
+                                                "class": "File",
+                                            }
                                         ),
-                                        {
-                                            "location": sf_location,
-                                            "basename": sfname,
-                                            "class": "File",
-                                        },
                                     )
                                 elif sf_required:
                                     raise SourceLine(
@@ -496,12 +491,10 @@ def addsf(
                                         debug,
                                     ).makeError(
                                         "Missing required secondary file '%s' from file object: %s"
-                                        % (sfname, json_dumps(datum, indent=4))
+                                        % (sfname, json_dumps(file_datum, indent=4))
                                     )
 
-                    normalizeFilesDirs(
-                        cast(MutableSequence[CWLObjectType], datum["secondaryFiles"])
-                    )
+                    normalizeFilesDirs(file_datum["secondaryFiles"])
 
                 if "format" in schema:
                     eval_format: Any = self.do_eval(schema["format"])
@@ -546,7 +539,7 @@ def addsf(
                         )
                     try:
                         check_format(
-                            datum,
+                            file_datum,
                             evaluated_format,
                             self.formatgraph,
                         )
@@ -557,21 +550,21 @@ def addsf(
                         ) from ve
 
                 visit_class(
-                    datum.get("secondaryFiles", []),
+                    file_datum.get("secondaryFiles", []),
                     ("File", "Directory"),
                     self._capture_files,
                 )
 
             if schema["type"] == "org.w3id.cwl.cwl.Directory":
-                datum = cast(CWLObjectType, datum)
+                dir_datum = cast(CWLDirectoryType, datum)
                 ll = schema.get("loadListing") or self.loadListing
                 if ll and ll != "no_listing":
                     get_listing(
                         self.fs_access,
-                        datum,
+                        dir_datum,
                         (ll == "deep_listing"),
                     )
-                self.files.append(datum)
+                self.files.append(dir_datum)
 
             if schema["type"] == "Any":
                 visit_class(datum, ("File", "Directory"), self._capture_files)
@@ -596,9 +589,7 @@ def tostr(self, value: MutableMapping[str, str] | Any) -> str:
         match value:
             case {"class": "File" | "Directory" as class_name, **rest}:
                 if "path" not in rest:
-                    raise WorkflowException(
-                        '{} object missing "path": {}'.format(class_name, value)
-                    )
+                    raise WorkflowException(f'{class_name} object missing "path": {value}')
                 return str(rest["path"])
             case ScalarFloat():
                 rep = RoundTripRepresenter()

cwltool/checker.py

@@ -3,14 +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, 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 CWLObjectType, CWLOutputType, SinkType, aslist
+from .utils import aslist
 
 
 def _get_type(tp: Any) -> Any:
@@ -21,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,
     valueFrom: str | None,
 ) -> Literal["pass"] | Literal["warning"] | Literal["exception"]:
@@ -55,7 +56,7 @@ 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():
@@ -66,7 +67,9 @@ def merge_flatten_type(src: SinkType) -> CWLOutputType:
             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.
 
@@ -84,8 +87,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(MutableSequence[CWLOutputType], src["items"]),
-                cast(MutableSequence[CWLOutputType], sink["items"]),
+                cast(MutableSequence[CWLOutputType | None], src["items"]),
+                cast(MutableSequence[CWLOutputType | None], sink["items"]),
                 strict,
             )
         if src["type"] == "record" and sink["type"] == "record":