freshen cwltool/update.py

Author: mr-c

cwltool/update.py

@@ -1,7 +1,6 @@
 import copy
 from functools import partial
 from typing import (
-    Any,
     Callable,
     Dict,
     List,
@@ -10,6 +9,7 @@
     Optional,
     Tuple,
     Union,
+    cast,
 )
 
 from ruamel.yaml.comments import CommentedMap, CommentedSeq
@@ -18,18 +18,19 @@
 from schema_salad.sourceline import SourceLine
 
 from .loghandler import _logger
-from .utils import aslist, visit_class, visit_field
+from .utils import CWLObjectType, CWLOutputType, aslist, visit_class, visit_field
 
 
-def v1_1to1_2(doc, loader, baseuri):  # pylint: disable=unused-argument
-    # type: (Any, Loader, str) -> Tuple[Any, str]
+def v1_1to1_2(
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     """Public updater for v1.1 to v1.2"""
 
     doc = copy.deepcopy(doc)
 
     upd = doc
     if isinstance(upd, MutableMapping) and "$graph" in upd:
-        upd = upd["$graph"]
+        upd = cast(CommentedMap, upd["$graph"])
     for proc in aslist(upd):
         if "cwlVersion" in proc:
             del proc["cwlVersion"]
@@ -38,8 +39,8 @@ def v1_1to1_2(doc, loader, baseuri):  # pylint: disable=unused-argument
 
 
 def v1_0to1_1(
-    doc: Any, loader: Loader, baseuri: str
-) -> Tuple[Any, str]:  # pylint: disable=unused-argument
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     """Public updater for v1.0 to v1.1."""
     doc = copy.deepcopy(doc)
 
@@ -52,29 +53,31 @@ def v1_0to1_1(
         "http://commonwl.org/cwltool#LoadListingRequirement": "LoadListingRequirement",
     }
 
-    def rewrite_requirements(t: MutableMapping[str, Any]) -> None:
+    def rewrite_requirements(t: CWLObjectType) -> None:
         if "requirements" in t:
-            for r in t["requirements"]:
+            for r in cast(MutableSequence[CWLObjectType], t["requirements"]):
                 if isinstance(r, MutableMapping):
-                    if r["class"] in rewrite:
-                        r["class"] = rewrite[r["class"]]
+                    cls = cast(str, r["class"])
+                    if cls in rewrite:
+                        r["class"] = rewrite[cls]
                 else:
                     raise ValidationException(
                         "requirements entries must be dictionaries: {} {}.".format(
                             type(r), r
                         )
                     )
         if "hints" in t:
-            for r in t["hints"]:
+            for r in cast(MutableSequence[CWLObjectType], t["hints"]):
                 if isinstance(r, MutableMapping):
-                    if r["class"] in rewrite:
-                        r["class"] = rewrite[r["class"]]
+                    cls = cast(str, r["class"])
+                    if cls in rewrite:
+                        r["class"] = rewrite[cls]
                 else:
                     raise ValidationException(
                         "hints entries must be dictionaries: {} {}.".format(type(r), r)
                     )
         if "steps" in t:
-            for s in t["steps"]:
+            for s in cast(MutableSequence[CWLObjectType], t["steps"]):
                 if isinstance(s, MutableMapping):
                     rewrite_requirements(s)
                 else:
@@ -83,25 +86,27 @@ def rewrite_requirements(t: MutableMapping[str, Any]) -> None:
                     )
 
     def update_secondaryFiles(t, top=False):
-        # type: (Any, bool) -> Union[MutableSequence[MutableMapping[str, str]], MutableMapping[str, str]]
+        # type: (CWLOutputType, bool) -> Union[MutableSequence[MutableMapping[str, str]], MutableMapping[str, str]]
         if isinstance(t, CommentedSeq):
             new_seq = copy.deepcopy(t)
             for index, entry in enumerate(t):
                 new_seq[index] = update_secondaryFiles(entry)
             return new_seq
         elif isinstance(t, MutableSequence):
-            return CommentedSeq([update_secondaryFiles(p) for p in t])
+            return CommentedSeq(
+                [update_secondaryFiles(cast(CWLOutputType, p)) for p in t]
+            )
         elif isinstance(t, MutableMapping):
-            return t
+            return cast(MutableMapping[str, str], t)
         elif top:
             return CommentedSeq([CommentedMap([("pattern", t)])])
         else:
             return CommentedMap([("pattern", t)])
 
-    def fix_inputBinding(t):  # type: (Dict[str, Any]) -> None
-        for i in t["inputs"]:
+    def fix_inputBinding(t: CWLObjectType) -> None:
+        for i in cast(MutableSequence[CWLObjectType], t["inputs"]):
             if "inputBinding" in i:
-                ib = i["inputBinding"]
+                ib = cast(CWLObjectType, i["inputBinding"])
                 for k in list(ib.keys()):
                     if k != "loadContents":
                         _logger.warning(
@@ -118,7 +123,7 @@ def fix_inputBinding(t):  # type: (Dict[str, Any]) -> None
 
     upd = doc
     if isinstance(upd, MutableMapping) and "$graph" in upd:
-        upd = upd["$graph"]
+        upd = cast(CommentedMap, upd["$graph"])
     for proc in aslist(upd):
         proc.setdefault("hints", CommentedSeq())
         proc["hints"].insert(
@@ -136,33 +141,35 @@ def fix_inputBinding(t):  # type: (Dict[str, Any]) -> None
     return (doc, "v1.1")
 
 
-def v1_1_0dev1to1_1(doc, loader, baseuri):  # pylint: disable=unused-argument
-    # type: (Any, Loader, str) -> Tuple[Any, str]
+def v1_1_0dev1to1_1(
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     return (doc, "v1.1")
 
 
-def v1_2_0dev1todev2(doc, loader, baseuri):  # pylint: disable=unused-argument
-    # type: (Any, Loader, str) -> Tuple[Any, str]
+def v1_2_0dev1todev2(
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     return (doc, "v1.2.0-dev2")
 
 
 def v1_2_0dev2todev3(
-    doc: Any, loader: Loader, baseuri: str
-) -> Tuple[Any, str]:  # pylint: disable=unused-argument
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     """Public updater for v1.2.0-dev2 to v1.2.0-dev3"""
     doc = copy.deepcopy(doc)
 
-    def update_pickvalue(t: Any) -> None:
-        for step in t["steps"]:
-            for inp in step["in"]:
+    def update_pickvalue(t: CWLObjectType) -> None:
+        for step in cast(MutableSequence[CWLObjectType], t["steps"]):
+            for inp in cast(MutableSequence[CWLObjectType], step["in"]):
                 if "pickValue" in inp:
                     if inp["pickValue"] == "only_non_null":
                         inp["pickValue"] = "the_only_non_null"
 
     visit_class(doc, "Workflow", update_pickvalue)
     upd = doc
     if isinstance(upd, MutableMapping) and "$graph" in upd:
-        upd = upd["$graph"]
+        upd = cast(CommentedMap, upd["$graph"])
     for proc in aslist(upd):
         if "cwlVersion" in proc:
             del proc["cwlVersion"]
@@ -172,14 +179,14 @@ def update_pickvalue(t: Any) -> None:
 UPDATES = {
     u"v1.0": v1_0to1_1,
     u"v1.1": v1_1to1_2,
-}  # type: Dict[str, Optional[Callable[[Any, Loader, str], Tuple[Any, str]]]]
+}  # type: Dict[str, Optional[Callable[[CommentedMap, Loader, str], Tuple[CommentedMap, str]]]]
 
 DEVUPDATES = {
     u"v1.1.0-dev1": v1_1_0dev1to1_1,
     u"v1.2.0-dev1": v1_2_0dev1todev2,
     u"v1.2.0-dev2": v1_2_0dev2todev3,
     u"v1.2.0-dev3": None,
-}  # type: Dict[str, Optional[Callable[[Any, Loader, str], Tuple[Any, str]]]]
+}  # type: Dict[str, Optional[Callable[[CommentedMap, Loader, str], Tuple[CommentedMap, str]]]]
 
 
 ALLUPDATES = UPDATES.copy()
@@ -190,18 +197,16 @@ def update_pickvalue(t: Any) -> None:
 ORIGINAL_CWLVERSION = "http://commonwl.org/cwltool#original_cwlVersion"
 
 
-def identity(doc, loader, baseuri):  # pylint: disable=unused-argument
-    # type: (Any, Loader, str) -> Tuple[Any, Union[str, str]]
+def identity(
+    doc: CommentedMap, loader: Loader, baseuri: str
+) -> Tuple[CommentedMap, str]:  # pylint: disable=unused-argument
     """Default, do-nothing, CWL document upgrade function."""
-    return (doc, doc["cwlVersion"])
+    return (doc, cast(str, doc["cwlVersion"]))
 
 
 def checkversion(
-    doc,  # type: Union[CommentedSeq, CommentedMap]
-    metadata,  # type: CommentedMap
-    enable_dev,  # type: bool
-):
-    # type: (...) -> Tuple[CommentedMap, str]
+    doc: Union[CommentedSeq, CommentedMap], metadata: CommentedMap, enable_dev: bool,
+) -> Tuple[CommentedMap, str]:
     """Check the validity of the version of the give CWL document.
 
     Returns the document and the validated version string.
@@ -251,15 +256,20 @@ def checkversion(
     return (cdoc, version)
 
 
-def update(doc, loader, baseuri, enable_dev, metadata):
-    # type: (Union[CommentedSeq, CommentedMap], Loader, str, bool, Any) -> CommentedMap
+def update(
+    doc: Union[CommentedSeq, CommentedMap],
+    loader: Loader,
+    baseuri: str,
+    enable_dev: bool,
+    metadata: CommentedMap,
+) -> CommentedMap:
 
     (cdoc, version) = checkversion(doc, metadata, enable_dev)
     originalversion = copy.copy(version)
 
     nextupdate = (
         identity
-    )  # type: Optional[Callable[[Any, Loader, str], Tuple[Any, str]]]
+    )  # type: Optional[Callable[[CommentedMap, Loader, str], Tuple[CommentedMap, str]]]
 
     while nextupdate:
         (cdoc, version) = nextupdate(cdoc, loader, baseuri)