types are hard

Author: mr-c

cwltool/command_line_tool.py

@@ -559,9 +559,9 @@ def calc_checksum(location: str) -> Optional[str]:
                 runtimeContext.outdir = jobcache
 
                 def update_status_output_callback(
-                    output_callbacks: Callable[[List[Dict[str, Any]], str], None],
+                    output_callbacks: OutputCallbackType,
                     jobcachelock: IO[Any],
-                    outputs: List[Dict[str, Any]],
+                    outputs: Optional[CWLObjectType],
                     processStatus: str,
                 ) -> None:
                     # save status to the lockfile then release the lock
@@ -624,44 +624,44 @@ def update_status_output_callback(
                 ls = builder.do_eval(initialWorkdir["listing"])
             else:
                 for t in cast(
-                    MutableSequence[Union[str, Dict[str, str]]],
+                    MutableSequence[Union[str, CWLObjectType]],
                     initialWorkdir["listing"],
                 ):
                     if isinstance(t, Mapping) and "entry" in t:
-                        entry_exp = builder.do_eval(t["entry"], strip_whitespace=False)
+                        entry_exp = builder.do_eval(cast(str, t["entry"]), strip_whitespace=False)
                         for entry in aslist(entry_exp):
                             et = {"entry": entry}
                             if "entryname" in t:
-                                et["entryname"] = builder.do_eval(t["entryname"])
+                                et["entryname"] = builder.do_eval(cast(str, t["entryname"]))
                             else:
                                 et["entryname"] = None
                             et["writable"] = t.get("writable", False)
                             if et["entry"] is not None:
                                 ls.append(et)
                     else:
-                        initwd_item = builder.do_eval(t)
+                        initwd_item = builder.do_eval(cast(str, t))
                         if not initwd_item:
                             continue
                         if isinstance(initwd_item, MutableSequence):
                             ls.extend(initwd_item)
                         else:
                             ls.append(initwd_item)
-            for i, t in enumerate(ls):
-                if "entry" in t:
-                    if isinstance(t["entry"], str):
+            for i, t2 in enumerate(ls):
+                if "entry" in t2:
+                    if isinstance(t2["entry"], str):
                         ls[i] = {
                             "class": "File",
-                            "basename": t["entryname"],
-                            "contents": t["entry"],
-                            "writable": t.get("writable"),
+                            "basename": t2["entryname"],
+                            "contents": t2["entry"],
+                            "writable": t2.get("writable"),
                         }
                     else:
-                        if t.get("entryname") or t.get("writable"):
-                            t = copy.deepcopy(t)
-                            if t.get("entryname"):
-                                t["entry"]["basename"] = t["entryname"]
-                            t["entry"]["writable"] = t.get("writable")
-                        ls[i] = t["entry"]
+                        if t2.get("entryname") or t2.get("writable"):
+                            t2 = copy.deepcopy(t2)
+                            if t2.get("entryname"):
+                                t2["entry"]["basename"] = t2["entryname"]
+                            t2["entry"]["writable"] = t2.get("writable")
+                        ls[i] = t2["entry"]
             j.generatefiles["listing"] = ls
             for entry in ls:
                 self.updatePathmap(builder.outdir, builder.pathmapper, entry)
@@ -787,8 +787,8 @@ def register_reader(f):  # type: (Dict[str, Any]) -> None
         j.environment = {}
         evr, _ = self.get_requirement("EnvVarRequirement")
         if evr is not None:
-            for t in cast(List[Dict[str, str]], evr["envDef"]):
-                j.environment[t["envName"]] = builder.do_eval(t["envValue"])
+            for t2 in cast(List[Dict[str, str]], evr["envDef"]):
+                j.environment[t2["envName"]] = builder.do_eval(t2["envValue"])
 
         shellcmd, _ = self.get_requirement("ShellCommandRequirement")
         if shellcmd is not None:

cwltool/provenance.py

@@ -1083,7 +1083,7 @@ def _initialize_bagit(self):  # type: () -> None
             bag_it_file.write("BagIt-Version: 0.97\n")
             bag_it_file.write("Tag-File-Character-Encoding: %s\n" % ENCODING)
 
-    def open_log_file_for_activity(self, uuid_uri: str) -> WritableBagFile:
+    def open_log_file_for_activity(self, uuid_uri: str) -> Union[TextIOWrapper, WritableBagFile]:
         self.self_check()
         # Ensure valid UUID for safe filenames
         activity_uuid = uuid.UUID(uuid_uri)
@@ -1140,8 +1140,7 @@ def user_provenance(self, document):  # type: (ProvDocument) -> None
         # get their name wrong!)
         document.actedOnBehalfOf(account, user)
 
-    def write_bag_file(self, path, encoding=ENCODING):
-        # type: (str, Optional[str]) -> WritableBagFile
+    def write_bag_file(self, path: str, encoding: Optional[str]=ENCODING)-> Union[TextIOWrapper, WritableBagFile]:
         """Write the bag file into our research object."""
         self.self_check()
         # For some reason below throws BlockingIOError
@@ -1150,12 +1149,9 @@ def write_bag_file(self, path, encoding=ENCODING):
         if encoding is not None:
             # encoding: match Tag-File-Character-Encoding: UTF-8
             # newline: ensure LF also on Windows
-            return cast(
-                WritableBagFile,
-                TextIOWrapper(
+            return TextIOWrapper(
                     cast(IO[bytes], bag_file), encoding=encoding, newline="\n"
-                ),
-            )
+                )
         return bag_file
 
     def add_tagfile(self, path, timestamp=None):
@@ -1503,8 +1499,7 @@ def packed_workflow(self, packed):  # type: (str) -> None
         rel_path = str(PurePosixPath(WORKFLOW) / "packed.cwl")
         # Write as binary
         with self.write_bag_file(rel_path, encoding=None) as write_pack:
-            # YAML is always UTF8, but json.dumps gives us str in py2
-            write_pack.write(packed.encode(ENCODING))
+            write_pack.write(packed)
         _logger.debug("[provenance] Added packed workflow: %s", rel_path)
 
     def has_data_file(self, sha1hash):  # type: (str) -> bool
@@ -1513,8 +1508,7 @@ def has_data_file(self, sha1hash):  # type: (str) -> bool
         hash_path = os.path.join(folder, sha1hash)
         return os.path.isfile(hash_path)
 
-    def add_data_file(self, from_fp, timestamp=None, content_type=None):
-        # type: (IO[Any], Optional[datetime.datetime], Optional[str]) -> str
+    def add_data_file(self, from_fp: IO[Any], timestamp: Optional[datetime.datetime]=None, content_type: Optional[str]=None) -> str:
         """Copy inputs to data/ folder."""
         self.self_check()
         tmp_dir, tmp_prefix = os.path.split(self.temp_prefix)
@@ -1554,17 +1548,15 @@ def add_data_file(self, from_fp, timestamp=None, content_type=None):
             self._content_types[rel_path] = content_type
         return rel_path
 
-    def _self_made(self, timestamp=None):
-        # type: (Optional[datetime.datetime]) -> Dict[str, Any]
+    def _self_made(self, timestamp: Optional[datetime.datetime] = None) -> Dict[str, Any]:
         if timestamp is None:
             timestamp = datetime.datetime.now()
         return {
             "createdOn": timestamp.isoformat(),
             "createdBy": {"uri": self.engine_uuid, "name": self.cwltool_version},
         }
 
-    def add_to_manifest(self, rel_path, checksums):
-        # type: (str, Dict[str,str]) -> None
+    def add_to_manifest(self, rel_path: str, checksums: Dict[str,str]) -> None:
         """Add files to the research object manifest."""
         self.self_check()
         if PurePosixPath(rel_path).is_absolute():
@@ -1593,8 +1585,7 @@ def add_to_manifest(self, rel_path, checksums):
                 _logger.debug("[provenance] Added to %s: %s", manifestpath, line)
                 checksum_file.write(line)
 
-    def _add_to_bagit(self, rel_path, **checksums):
-        # type: (str, Any) -> None
+    def _add_to_bagit(self, rel_path: str, **checksums: str) -> None:
         if PurePosixPath(rel_path).is_absolute():
             raise ValueError("rel_path must be relative: %s" % rel_path)
         local_path = os.path.join(self.folder, _local_path(rel_path))
@@ -1714,8 +1705,7 @@ def _relativise_files(
         except TypeError:
             pass
 
-    def close(self, save_to=None):
-        # type: (Optional[str]) -> None
+    def close(self, save_to: Optional[str]=None) -> None:
         """Close the Research Object, optionally saving to specified folder.
 
         Closing will remove any temporary files used by this research object.
@@ -1748,11 +1738,11 @@ def close(self, save_to=None):
 
 
 def checksum_copy(
-    src_file,  # type: IO[Any]
-    dst_file=None,  # type: Optional[IO[Any]]
-    hasher=Hasher,  # type: Callable[[], hashlib._Hash]
-    buffersize=1024 * 1024,  # type: int
-):  # type: (...) -> str
+    src_file: IO[Any],
+    dst_file: Optional[IO[Any]] = None,
+    hasher = Hasher,  # type: Callable[[], hashlib._Hash]
+    buffersize: int=1024 * 1024,
+) -> str:
     """Compute checksums while copying a file."""
     # TODO: Use hashlib.new(Hasher_str) instead?
     checksum = hasher()

cwltool/workflow.py

@@ -448,10 +448,11 @@ def postScatterEval(io: CWLObjectType) -> Optional[CWLObjectType]:
 
                 fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)("")
                 for k, v in io.items():
-                    val = cast(Dict[str, Any], v)
-                    if k in loadContents and val.get("contents") is None:
-                        with fs_access.open(val["location"], "rb") as f:
-                            val["contents"] = content_limit_respected_read(f)
+                    if k in loadContents:
+                        val = cast(Dict[str, Any], v)
+                        if val.get("contents") is None:
+                            with fs_access.open(val["location"], "rb") as f:
+                                val["contents"] = content_limit_respected_read(f)
 
                 def valueFromFunc(k: Any, v: Any) -> Any:
                     if k in valueFrom: