upgrade to mypy 0.971 (#139)

Author: mr-c

cwl_utils/cwl_normalizer.py

@@ -83,9 +83,7 @@ def run(args: argparse.Namespace) -> int:
         add_lc_filename(result, document)
         version = result.get("cwlVersion", None)
         if version in ("draft-3", "cwl:draft-3", "v1.0", "v1.1"):
-            result = cwlupgrader.upgrade_document(
-                result, False, False, args.dir, imports
-            )
+            result = cwlupgrader.upgrade_document(result, args.dir, imports=imports)
         else:
             _logger.error(
                 "Sorry, %s in %s is not a supported CWL version by this tool.",

cwl_utils/graph_split.py

@@ -11,10 +11,12 @@
 import argparse
 import json
 import os
-from typing import Any, IO, MutableMapping, Set, Union, cast
+from typing import IO, Any, MutableMapping, Set, Union, cast
 
 from cwlformat.formatter import stringify_dict
-from ruamel import yaml
+from ruamel.yaml.main import YAML, dump
+from ruamel.yaml.representer import RoundTripRepresenter
+from ruamel.yaml.dumper import RoundTripDumper
 from schema_salad.sourceline import SourceLine, add_lc_filename
 
 
@@ -72,7 +74,9 @@ def run(
     sourceIO: IO[str], output_dir: str, output_format: str, mainfile: str, pretty: bool
 ) -> None:
     """Loop over the provided packed CWL document and split it up."""
-    source = yaml.main.round_trip_load(sourceIO, preserve_quotes=True)
+    yaml = YAML(typ="rt")
+    yaml.preserve_quotes = True  # type: ignore[assignment]
+    source = yaml.load(sourceIO)
     add_lc_filename(source, sourceIO.name)
 
     if "$graph" not in source:
@@ -87,7 +91,7 @@ def my_represent_none(
         """Force clean representation of 'null'."""
         return self.represent_scalar("tag:yaml.org,2002:null", "null")
 
-    yaml.representer.RoundTripRepresenter.add_representer(type(None), my_represent_none)
+    RoundTripRepresenter.add_representer(type(None), my_represent_none)
 
     for entry in source["$graph"]:
         entry_id = entry.pop("id")[1:]
@@ -205,9 +209,7 @@ def rewrite_schemadef(document: MutableMapping[str, Any]) -> Set[str]:
                 field["name"] = field["name"].split("/")[2]
                 rewrite_types(field, entry_file, True)
             with open(entry_file[1:], "a", encoding="utf-8") as entry_handle:
-                yaml.main.dump(
-                    [entry], entry_handle, Dumper=yaml.dumper.RoundTripDumper
-                )
+                dump([entry], entry_handle, Dumper=RoundTripDumper)
             entry["$import"] = entry_file[1:]
             del entry["name"]
             del entry["type"]
@@ -236,16 +238,17 @@ def json_dump(entry: Any, output_file: str) -> None:
 
 def yaml_dump(entry: Any, output_file: str, pretty: bool) -> None:
     """Output object as YAML."""
+    yaml = YAML(typ="rt")
+    yaml.default_flow_style = False
+    yaml.map_indent = 4  # type: ignore[assignment]
+    yaml.sequence_indent = 2  # type: ignore[assignment]
     with open(output_file, "w", encoding="utf-8") as result_handle:
         if pretty:
             result_handle.write(stringify_dict(entry))
         else:
-            yaml.main.round_trip_dump(
+            yaml.dump(
                 entry,
                 result_handle,
-                default_flow_style=False,
-                indent=4,
-                block_seq_indent=2,
             )
 
 

cwl_utils/pack.py

@@ -18,12 +18,25 @@
 import sys
 import urllib.parse
 import urllib.request
-from typing import Any, Dict, ItemsView, List, Optional, Tuple, cast
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    ItemsView,
+    List,
+    Optional,
+    Tuple,
+    Union,
+    cast,
+)
 
 from packaging import version
 
 from cwl_utils import schemadef, utils
 
+if TYPE_CHECKING:
+    from _collections_abc import dict_items
+
 logger = logging.getLogger(__name__)
 
 
@@ -167,7 +180,7 @@ def resolve_schemadefs(
 
 def resolve_imports(cwl: Any, base_url: urllib.parse.ParseResult) -> Any:
     if isinstance(cwl, dict):
-        itr = cwl.items()
+        itr: Union["dict_items[Any, Any]", ItemsView[Any, Any]] = cwl.items()
     elif isinstance(cwl, list):
         itr = cast(ItemsView[Any, Any], [(n, v) for n, v in enumerate(cwl)])
     else:

cwl_utils/sandboxjs.py

@@ -97,12 +97,13 @@ def __init__(
         self.have_node_slim: bool = have_node_slim
         self.minimum_node_version_str: str = minimum_node_version_str
         self.process_finished_str: str = process_finished_str
-        self.processes_to_kill = (
-            collections.deque()
-        )  # type: Deque[subprocess.Popen[str]]
+        self.processes_to_kill: Deque[subprocess.Popen[str]] = collections.deque()
 
     def __del__(self) -> None:
-        kill_processes(self.processes_to_kill)
+        try:
+            kill_processes(self.processes_to_kill)
+        except TypeError:
+            pass
 
     def check_js_threshold_version(self, working_alias: str) -> bool:
         """

cwl_utils/utils.py

@@ -51,8 +51,7 @@ def _is_github_symbolic_link(base_url: urllib.parse.ParseResult, contents: str)
 
 def bytes2str_in_dicts(
     inp: Union[MutableMapping[str, Any], MutableSequence[Any], Any],
-):
-    # type: (...) -> Union[str, MutableSequence[Any], MutableMapping[str, Any]]
+) -> Union[str, MutableSequence[Any], MutableMapping[str, Any]]:
     """
     Convert any present byte string to unicode string, inplace.
 

mypy-requirements.txt

@@ -1,4 +1,4 @@
-mypy==0.910
+mypy==0.971
 typing_extensions
 types-requests
 types-setuptools>==57.4.0

requirements.txt

@@ -1,4 +1,4 @@
-cwl-upgrader >= 1.2
+cwl-upgrader >= 1.2.3
 packaging
 requests
 schema-salad >= 8.2, < 9

tests/load_cwl_by_path.py

@@ -2,7 +2,6 @@
 # SPDX-License-Identifier: Apache-2.0
 from pathlib import Path
 
-
 from cwl_utils.parser import load_document_by_uri, save
 
 # File Input - This is the only thing you will need to adjust or take in as an input to your function:

tests/test_parser.py

@@ -1,7 +1,7 @@
 """Test the load and save functions for CWL."""
 from pathlib import Path
 
-from ruamel import yaml
+from ruamel.yaml.main import YAML
 
 import cwl_utils.parser.latest as latest
 from cwl_utils.parser import (
@@ -16,20 +16,22 @@
 TEST_v1_0_CWL = HERE / "../testdata/md5sum.cwl"
 TEST_v1_0_CWL_REMOTE = "https://raw.githubusercontent.com/common-workflow-language/cwl-utils/main/testdata/md5sum.cwl"
 TEST_v1_2_CWL = HERE / "../testdata/workflow_input_format_expr_v1_2.cwl"
+yaml = YAML(typ="rt")
+yaml.preserve_quotes = True  # type: ignore[assignment]
 
 
 def test_cwl_version() -> None:
     """Test cwl_version for a CommandLineTool."""
     with open(TEST_v1_0_CWL) as cwl_h:
-        yaml_obj = yaml.main.round_trip_load(cwl_h, preserve_quotes=True)
+        yaml_obj = yaml.load(cwl_h)
     ver = cwl_version(yaml_obj)
     assert ver == "v1.0"
 
 
 def test_load_document() -> None:
     """Test load_document for a CommandLineTool."""
     with open(TEST_v1_0_CWL) as cwl_h:
-        yaml_obj = yaml.main.round_trip_load(cwl_h, preserve_quotes=True)
+        yaml_obj = yaml.load(cwl_h)
     cwl_obj = load_document(yaml_obj)
     assert cwl_obj.cwlVersion == "v1.0"
     assert cwl_obj.inputs[0].id.endswith("input_file")
@@ -54,12 +56,12 @@ def test_load_document_with_remote_uri() -> None:
 def test_save() -> None:
     """Test save for a list of Process objects with different cwlVersions."""
     with open(TEST_v1_0_CWL) as cwl_h:
-        yaml_obj10 = yaml.main.round_trip_load(cwl_h, preserve_quotes=True)
+        yaml_obj10 = yaml.load(cwl_h)
     cwl_obj10 = load_document(yaml_obj10)
     assert cwl_obj10.cwlVersion == "v1.0"
 
     with open(TEST_v1_2_CWL) as cwl_h:
-        yaml_obj12 = yaml.main.round_trip_load(cwl_h, preserve_quotes=True)
+        yaml_obj12 = yaml.load(cwl_h)
     cwl_obj12 = load_document(yaml_obj12)
     assert cwl_obj12.cwlVersion == "v1.2"
 
@@ -72,8 +74,8 @@ def test_latest_parser() -> None:
     """Test the `latest` parser is same as cwl_v1_2 (current latest) parser."""
     uri = Path(TEST_v1_2_CWL).as_uri()
     with open(TEST_v1_2_CWL) as cwl_h:
-        yaml_obj12 = yaml.main.round_trip_load(cwl_h, preserve_quotes=True)
-    latest_cwl_obj = latest.load_document_by_yaml(yaml_obj12, uri)  # type: ignore
+        yaml_obj12 = yaml.load(cwl_h)
+    latest_cwl_obj = latest.load_document_by_yaml(yaml_obj12, uri)
     assert latest_cwl_obj.cwlVersion == "v1.2"