Apply IsType optimizations

Author: GlassOfWhiskey

cwltool/command_line_tool.py

@@ -14,15 +14,14 @@
 import urllib.parse
 from collections.abc import (
     Generator,
-    Iterable,
     Mapping,
     MutableMapping,
     MutableSequence,
 )
 from enum import Enum
 from functools import cmp_to_key, partial
 from re import Pattern
-from typing import TYPE_CHECKING, Any, Optional, TextIO, Union, cast
+from typing import Any, Optional, TYPE_CHECKING, TextIO, Union, cast
 
 from cwl_utils.types import (
     CWLDirectoryType,
@@ -42,12 +41,7 @@
 from schema_salad.utils import json_dumps
 from schema_salad.validate import validate_ex
 
-from .builder import (
-    INPUT_OBJ_VOCAB,
-    Builder,
-    content_limit_respected_read_bytes,
-    substitute,
-)
+from .builder import Builder, INPUT_OBJ_VOCAB, content_limit_respected_read_bytes, substitute
 from .context import LoadingContext, RuntimeContext, getdefault
 from .docker import DockerCommandLineJob, PodmanCommandLineJob
 from .errors import UnsupportedRequirement, WorkflowException
@@ -1292,7 +1286,7 @@ def collect_output(
         fs_access: StdFsAccess,
         compute_checksum: bool = True,
     ) -> CWLOutputType | None:
-        r: list[CWLOutputType] = []
+        r: MutableSequence[CWLFileType | CWLDirectoryType] = []
         empty_and_optional = False
         debug = _logger.isEnabledFor(logging.DEBUG)
         result: CWLOutputType | None = None
@@ -1340,9 +1334,9 @@ def collect_output(
                                 key=cmp_to_key(locale.strcoll),
                             )
                             r.extend(
-                                cast(
-                                    Iterable[CWLOutputType],
-                                    [
+                                [
+                                    cast(
+                                        CWLFileType | CWLDirectoryType,
                                         {
                                             "location": g,
                                             "path": fs_access.join(
@@ -1353,31 +1347,31 @@ def collect_output(
                                             "nameroot": os.path.splitext(decoded_basename)[0],
                                             "nameext": os.path.splitext(decoded_basename)[1],
                                             "class": "File" if fs_access.isfile(g) else "Directory",
-                                        }
-                                        for g, decoded_basename in zip(
+                                        },
+                                    )
+                                    for g, decoded_basename in zip(
+                                        sorted_glob_result,
+                                        map(
+                                            lambda x: os.path.basename(urllib.parse.unquote(x)),
                                             sorted_glob_result,
-                                            map(
-                                                lambda x: os.path.basename(urllib.parse.unquote(x)),
-                                                sorted_glob_result,
-                                            ),
-                                        )
-                                    ],
-                                )
+                                        ),
+                                    )
+                                ]
                             )
                         except OSError as e:
                             _logger.warning(str(e), exc_info=builder.debug)
                         except Exception:
                             _logger.error("Unexpected error from fs_access", exc_info=True)
                             raise
 
-                for files in cast(MutableSequence[CWLFileType | CWLDirectoryType], r):
+                for files in r:
                     rfile = files.copy()
                     revmap(rfile)
                     if is_directory(files):
                         ll = binding.get("loadListing") or builder.loadListing
                         if ll and ll != "no_listing":
                             get_listing(fs_access, files, (ll == "deep_listing"))
-                    elif is_file(files):
+                    else:
                         if binding.get("loadContents"):
                             with fs_access.open(rfile["location"], "rb") as f:
                                 files["contents"] = str(

cwltool/cwlprov/provenance_profile.py

@@ -312,7 +312,7 @@ def declare_file(self, value: _CWLFileArtifact) -> tuple[ProvEntity, ProvEntity,
         self.document.specializationOf(file_entity, entity)
 
         # Check for secondaries
-        for sec in cast(MutableSequence[CWLObjectType], value.get("secondaryFiles", [])):
+        for sec in value.get("secondaryFiles", []):
             # TODO: Record these in a specializationOf entity with UUID?
             if is_file(sec):
                 (sec_entity, _, _) = self.declare_file(cast(_CWLFileArtifact, sec))

cwltool/pathmapper.py

@@ -6,7 +6,7 @@
 from collections.abc import ItemsView, Iterable, Iterator, KeysView, MutableSequence
 from typing import NamedTuple, Optional, cast
 
-from cwl_utils.types import CWLDirectoryType, CWLFileType, is_directory, is_file
+from cwl_utils.types import CWLDirectoryType, CWLFileType, is_directory
 from mypy_extensions import mypyc_attr
 from schema_salad.exceptions import ValidationException
 from schema_salad.ref_resolver import uri_file_path
@@ -137,7 +137,7 @@ def visit(
                 copy=copy,
                 staged=staged,
             )
-        elif is_file(obj):
+        else:
             path = obj["location"]
             ab = abspath(path, basedir)
             if "contents" in obj and path.startswith("_:"):

cwltool/process.py

@@ -1259,7 +1259,7 @@ def scandeps(
                         }
                     )
                     if "listing" in doc:
-                        deps["listing"] = cast(CWLDirectoryType, doc)["listing"]
+                        deps["listing"] = doc["listing"]
                 deps["location"] = urljoin(base, u)
                 if "basename" in doc:
                     deps["basename"] = doc["basename"]