Solving some typing problems

Author: GlassOfWhiskey

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 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,

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:

cwltool/command_line_tool.py

@@ -24,6 +24,14 @@
 from re import Pattern
 from typing import TYPE_CHECKING, Any, Optional, TextIO, Union, cast
 
+from cwl_utils.types import (
+    CWLObjectType,
+    CWLOutputType,
+    DirectoryType,
+    is_directory,
+    is_file,
+    is_file_or_directory,
+)
 from mypy_extensions import mypyc_attr
 from ruamel.yaml.comments import CommentedMap, CommentedSeq
 from schema_salad.avro.schema import RecordSchema
@@ -60,9 +68,6 @@
 from .udocker import UDockerCommandLineJob
 from .update import ORDERED_VERSIONS, ORIGINAL_CWLVERSION
 from .utils import (
-    CWLObjectType,
-    CWLOutputType,
-    DirectoryType,
     JobsGeneratorType,
     OutputCallbackType,
     adjustDirObjs,
@@ -619,11 +624,8 @@ def _initialworkdir(self, j: JobBase | None, builder: Builder) -> None:
                     et: CWLObjectType = {}
                     writable = t.get("writable", False)
                     et["writable"] = writable
-                    if isinstance(entry, Mapping) and entry.get("class") in (
-                        "File",
-                        "Directory",
-                    ):
-                        if writable and "secondaryFiles" in entry:
+                    if is_file_or_directory(entry):
+                        if writable and is_file(entry) and "secondaryFiles" in entry:
                             secFiles = cast(MutableSequence[CWLObjectType], entry["secondaryFiles"])
                             for sf in secFiles:
                                 sf["writable"] = writable
@@ -765,22 +767,23 @@ def _initialworkdir(self, j: JobBase | None, builder: Builder) -> None:
             for entry in ls:
                 if "basename" in entry:
                     basename = cast(str, entry["basename"])
-                    dirname = os.path.join(builder.outdir, os.path.dirname(basename))
-                    entry["dirname"] = dirname
                     entry["basename"] = os.path.basename(basename)
-                    if "secondaryFiles" in entry:
-                        for sec_file in cast(
-                            MutableSequence[CWLObjectType], entry["secondaryFiles"]
-                        ):
-                            sec_file["dirname"] = dirname
+                    if is_file(entry):
+                        dirname = os.path.join(builder.outdir, os.path.dirname(basename))
+                        entry["dirname"] = dirname
+                        if "secondaryFiles" in entry:
+                            for sec_file in cast(
+                                MutableSequence[CWLObjectType], entry["secondaryFiles"]
+                            ):
+                                sec_file["dirname"] = dirname
                 normalizeFilesDirs(entry)
                 self.updatePathmap(
-                    cast(Optional[str], entry.get("dirname")) or builder.outdir,
+                    (entry.get("dirname") if is_file(entry) else None) or builder.outdir,
                     cast(PathMapper, builder.pathmapper),
                     entry,
                 )
 
-                if "listing" in entry:
+                if is_directory(entry) and "listing" in entry:
 
                     def remove_dirname(d: CWLObjectType) -> None:
                         if "dirname" in d:

cwltool/context.py

@@ -8,6 +8,7 @@
 from collections.abc import Callable, Iterable
 from typing import IO, TYPE_CHECKING, Any, Literal, Optional, TextIO, Union
 
+from cwl_utils.types import CWLObjectType
 from ruamel.yaml.comments import CommentedMap
 from schema_salad.avro.schema import Names
 from schema_salad.ref_resolver import Loader
@@ -16,7 +17,7 @@
 from .mpi import MpiConfig
 from .pathmapper import PathMapper
 from .stdfsaccess import StdFsAccess
-from .utils import DEFAULT_TMP_PREFIX, CWLObjectType, HasReqsHints, ResolverType
+from .utils import DEFAULT_TMP_PREFIX, HasReqsHints, ResolverType
 
 if TYPE_CHECKING:
     from _typeshed import SupportsWrite

cwltool/cuda.py

@@ -3,8 +3,9 @@
 import subprocess  # nosec
 import xml.dom.minidom  # nosec
 
+from cwl_utils.types import CWLObjectType
+
 from .loghandler import _logger
-from .utils import CWLObjectType
 
 
 def cuda_version_and_device_count() -> tuple[str, int]:

cwltool/cwlprov/provenance_profile.py

@@ -8,6 +8,7 @@
 from pathlib import PurePath, PurePosixPath
 from typing import TYPE_CHECKING, Any, cast
 
+from cwl_utils.types import CWLObjectType
 from prov.identifier import Identifier, QualifiedName
 from prov.model import PROV, PROV_LABEL, PROV_TYPE, PROV_VALUE, ProvDocument, ProvEntity
 from schema_salad.sourceline import SourceLine
@@ -17,7 +18,7 @@
 from ..loghandler import _logger
 from ..process import Process, shortname
 from ..stdfsaccess import StdFsAccess
-from ..utils import CWLObjectType, JobsType, get_listing, posix_path, versionstring
+from ..utils import JobsType, get_listing, posix_path, versionstring
 from ..workflow_job import WorkflowJob
 from .provenance_constants import (
     ACCOUNT_UUID,

cwltool/cwlprov/ro.py

@@ -13,13 +13,12 @@
 from typing import IO, TYPE_CHECKING, Any, Optional, cast
 
 import prov.model as provM
+from cwl_utils.types import CWLObjectType, CWLOutputType, is_directory, is_file
 from prov.model import ProvDocument
 
 from ..loghandler import _logger
 from ..stdfsaccess import StdFsAccess
 from ..utils import (
-    CWLObjectType,
-    CWLOutputType,
     create_tmp_dir,
     local_path,
     posix_path,
@@ -642,10 +641,10 @@ def _relativise_files(
         _logger.debug("[provenance] Relativising: %s", structure)
 
         if isinstance(structure, MutableMapping):
-            if structure.get("class") == "File":
+            if is_file(structure):
                 relative_path: str | PurePosixPath | None = None
                 if "checksum" in structure:
-                    raw_checksum = cast(str, structure["checksum"])
+                    raw_checksum = structure["checksum"]
                     alg, checksum = raw_checksum.split("$")
                     if alg != SHA1:
                         raise TypeError(
@@ -659,7 +658,7 @@ def _relativise_files(
                     # Register in RO; but why was this not picked
                     # up by used_artefacts?
                     _logger.info("[provenance] Adding to RO %s", structure["location"])
-                    with self.fsaccess.open(cast(str, structure["location"]), "rb") as fp:
+                    with self.fsaccess.open(structure["location"], "rb") as fp:
                         relative_path = self.add_data_file(fp)
                         checksum = PurePosixPath(relative_path).name
                         structure["checksum"] = f"{SHA1}${checksum}"
@@ -668,7 +667,7 @@ def _relativise_files(
                 if "path" in structure:
                     del structure["path"]
 
-            if structure.get("class") == "Directory":
+            if is_directory(structure):
                 # TODO: Generate anonymous Directory with a "listing"
                 # pointing to the hashed files
                 del structure["location"]

cwltool/cwlprov/writablebagfile.py

@@ -14,10 +14,11 @@
 from pathlib import Path, PurePosixPath
 from typing import Any, BinaryIO, cast
 
+from cwl_utils.types import CWLObjectType
 from schema_salad.utils import json_dumps
 
 from ..loghandler import _logger
-from ..utils import CWLObjectType, local_path, posix_path
+from ..utils import local_path, posix_path
 from .provenance_constants import (
     CWLPROV,
     CWLPROV_VERSION,

cwltool/docker.py

@@ -14,6 +14,7 @@
 from typing import Optional, cast
 
 import requests
+from cwl_utils.types import CWLObjectType
 
 from .builder import Builder
 from .context import RuntimeContext
@@ -22,7 +23,7 @@
 from .job import ContainerCommandLineJob
 from .loghandler import _logger
 from .pathmapper import MapperEnt, PathMapper
-from .utils import CWLObjectType, create_tmp_dir, ensure_writable
+from .utils import create_tmp_dir, ensure_writable
 
 _IMAGES: set[str] = set()
 _IMAGES_LOCK = threading.Lock()

cwltool/executors.py

@@ -13,6 +13,7 @@
 from typing import Optional, cast
 
 import psutil
+from cwl_utils.types import CWLObjectType
 from mypy_extensions import mypyc_attr
 from schema_salad.exceptions import ValidationException
 from schema_salad.sourceline import SourceLine
@@ -27,7 +28,7 @@
 from .process import Process, cleanIntermediate, relocateOutputs
 from .task_queue import TaskQueue
 from .update import ORIGINAL_CWLVERSION
-from .utils import CWLObjectType, JobsType
+from .utils import JobsType
 from .workflow import Workflow
 from .workflow_job import WorkflowJob, WorkflowJobStep