bootstrap-traverser.py: Add type hints and fix style

For maximum compatibility, we use the uppercase types from the
typing package instead of the built-in types, therefore compliant
with PEP 484 and PEP 526.

1 files changed, 53 insertions, 33 deletions

diff --git a/bin/bootstrap-traverser.py b/bin/bootstrap-traverser.py
index a58abb04..46388cde 100755
--- a/bin/bootstrap-traverser.py
+++ b/bin/bootstrap-traverser.py
@@ -13,41 +13,50 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import hashlib
 import json
 import os
-import shutil
 import subprocess
 import sys
+from typing import Any, Dict, List, Optional, cast
 
 from argparse import ArgumentParser
 
-def log(*args, **kwargs):
+# generic JSON type that avoids getter issues; proper use is being enforced by
+# return types of methods and typing vars holding return values of json getter
+Json = Dict[str, Any]
+
+
+def log(*args: str, **kwargs: Any) -> None:
     print(*args, file=sys.stderr, **kwargs)
 
-def fail(s):
+
+def fail(s: str) -> None:
     log(s)
     sys.exit(1)
 
-def git_hash(content):
-  header = "blob {}\0".format(len(content)).encode('utf-8')
-  h = hashlib.sha1()
-  h.update(header)
-  h.update(content)
-  return h.hexdigest()
 
-def create_blobs(blobs, *, root):
+def git_hash(content: bytes) -> str:
+    header = "blob {}\0".format(len(content)).encode('utf-8')
+    h = hashlib.sha1()
+    h.update(header)
+    h.update(content)
+    return h.hexdigest()
+
+
+def create_blobs(blobs: List[str], *, root: str) -> None:
     os.makedirs(os.path.join(root, "KNOWN"))
     for blob in blobs:
         blob_bin = blob.encode('utf-8')
         with open(os.path.join(root, "KNOWN", git_hash(blob_bin)), "wb") as f:
             f.write(blob_bin)
 
-def build_known(desc, *, root):
+
+def build_known(desc: Json, *, root: str) -> str:
     return os.path.join(root, "KNOWN", desc["data"]["id"])
 
-def link(src, dest):
+
+def link(src: str, dest: str) -> None:
     dest = os.path.normpath(dest)
     os.makedirs(os.path.dirname(dest), exist_ok=True)
     try:
@@ -55,51 +64,58 @@ def link(src, dest):
     except:
         os.symlink(src, dest)
 
-def build_local(desc, *, root, config):
+
+def build_local(desc: Json, *, root: str, config: Json) -> Optional[str]:
     repo_name = desc["data"]["repository"]
-    repo = config["repositories"][repo_name]["workspace_root"]
+    repo: List[str] = config["repositories"][repo_name]["workspace_root"]
     rel_path = desc["data"]["path"]
     if repo[0] == "file":
         return os.path.join(repo[1], rel_path)
-    fail("Unsupported repository root %r" % (repo,))
+    fail("Unsupported repository root %r" % (repo, ))
+
 
-def build_tree(desc, *, config, root, graph):
+def build_tree(desc: Json, *, config: Json, root: str, graph: Json) -> str:
     tree_id = desc["data"]["id"]
     tree_dir = os.path.normpath(os.path.join(root, "TREE", tree_id))
     if os.path.isdir(tree_dir):
         return tree_dir
     tree_desc = graph["trees"][tree_id]
     for location, desc in tree_desc.items():
-        link(build(desc, config=config, root=root, graph=graph),
+        link(cast(str, build(desc, config=config, root=root, graph=graph)),
              os.path.join(tree_dir, location))
     # correctly handle the empty tree
     os.makedirs(tree_dir, exist_ok=True)
     return tree_dir
 
-def run_action(action_id, *, config, root, graph):
+
+def run_action(action_id: str, *, config: Json, root: str, graph: Json) -> str:
     action_dir = os.path.normpath(os.path.join(root, "ACTION", action_id))
     if os.path.isdir(action_dir):
         return action_dir
     os.makedirs(action_dir)
     action_desc = graph["actions"][action_id]
     for location, desc in action_desc.get("input", {}).items():
-        link(build(desc, config=config, root=root, graph=graph),
+        link(cast(str, build(desc, config=config, root=root, graph=graph)),
              os.path.join(action_dir, location))
     cmd = action_desc["command"]
     env = action_desc.get("env")
-    log("Running %r with env %r for action %r"
-        % (cmd, env, action_id))
+    log("Running %r with env %r for action %r" % (cmd, env, action_id))
     for out in action_desc["output"]:
         os.makedirs(os.path.join(action_dir, os.path.dirname(out)),
                     exist_ok=True)
     subprocess.run(cmd, env=env, cwd=action_dir, check=True)
     return action_dir
 
-def build_action(desc, *, config, root, graph):
-    action_dir = run_action(desc["data"]["id"], config=config, root=root, graph=graph)
+
+def build_action(desc: Json, *, config: Json, root: str, graph: Json) -> str:
+    action_dir = run_action(desc["data"]["id"],
+                            config=config,
+                            root=root,
+                            graph=graph)
     return os.path.join(action_dir, desc["data"]["path"])
 
-def build(desc, *, config, root, graph):
+
+def build(desc: Json, *, config: Json, root: str, graph: Json) -> Optional[str]:
     if desc["type"] == "TREE":
         return build_tree(desc, config=config, root=root, graph=graph)
     if desc["type"] == "ACTION":
@@ -108,16 +124,19 @@ def build(desc, *, config, root, graph):
         return build_known(desc, root=root)
     if desc["type"] == "LOCAL":
         return build_local(desc, root=root, config=config)
-    fail("Don't know how to build artifact %r" % (desc,))
+    fail("Don't know how to build artifact %r" % (desc, ))
+
 
-def traverse(*, graph, to_build, out, root, config):
+def traverse(*, graph: Json, to_build: Json, out: str, root: str,
+             config: Json) -> None:
     os.makedirs(out, exist_ok=True)
     os.makedirs(root, exist_ok=True)
     create_blobs(graph["blobs"], root=root)
     for location, artifact in to_build.items():
-        link(build(artifact, config=config, root=root, graph=graph),
+        link(cast(str, build(artifact, config=config, root=root, graph=graph)),
              os.path.join(out, location))
 
+
 def main():
     parser = ArgumentParser()
     parser.add_argument("-C",
@@ -138,14 +157,13 @@ def main():
 
     (options, args) = parser.parse_known_args()
     if len(args) != 2:
-        fail("usage: %r <graph> <targets_to_build>"
-             % (sys.argv[0],))
+        fail("usage: %r <graph> <targets_to_build>" % (sys.argv[0], ))
     with open(args[0]) as f:
         graph = json.load(f)
     with open(args[1]) as f:
         to_build = json.load(f)
-    out = os.path.abspath(options.output_directory or "out-boot")
-    root = os.path.abspath(options.local_build_root or ".just-boot")
+    out = os.path.abspath(cast(str, options.output_directory or "out-boot"))
+    root = os.path.abspath(cast(str, options.local_build_root or ".just-boot"))
     with open(options.repository_config or "repo-conf.json") as f:
         config = json.load(f)
     if options.default_workspace:
@@ -153,7 +171,9 @@ def main():
         repos = config.get("repositories", {}).keys()
         for repo in repos:
             if not "workspace_root" in config["repositories"][repo]:
-                config["repositories"][repo]["workspace_root"] = ["file", ws_root]
+                config["repositories"][repo]["workspace_root"] = [
+                    "file", ws_root
+                ]
     traverse(graph=graph, to_build=to_build, out=out, root=root, config=config)