Merge pull request #1285 from adRn-s/egd2022_typehints

Type annotations for planemo.bioblend, planemo.git, and planemo.cwl.run

planemo/bioblend.py

@@ -15,7 +15,7 @@
 )
 
 
-def ensure_module():
+def ensure_module() -> None:
     """Throw an exception if bioblend is not available to Planemo."""
     if toolshed is None:
         raise Exception(BIOBLEND_UNAVAILABLE)

planemo/cwl/run.py

@@ -6,6 +6,13 @@
 """
 import json
 import tempfile
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    TYPE_CHECKING,
+    Union,
+)
 
 from galaxy.tool_util.cwl.cwltool_deps import (
     ensure_cwltool_available,
@@ -19,38 +26,49 @@
 )
 from planemo.runnable import (
     ErrorRunResponse,
+    Runnable,
     SuccessfulRunResponse,
 )
 
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
+
 JSON_PARSE_ERROR_MESSAGE = "Failed to parse JSON from cwltool output [%s] " "in file [%s]. cwltool logs [%s]."
 
 
 class CwlToolRunResponse(SuccessfulRunResponse):
     """Describe the resut of a cwltool invocation."""
 
-    def __init__(self, runnable, log, outputs=None):
+    def __init__(
+        self,
+        runnable: "Runnable",
+        log: str,
+        outputs: Optional[Dict[str, Any]] = None,
+    ) -> None:
         self._runnable = runnable
         self._log = log
         self._outputs = outputs
 
     @property
-    def log(self):
+    def log(self) -> str:
         return self._log
 
     @property
-    def job_info(self):
+    def job_info(self) -> None:
         return None
 
     @property
-    def invocation_details(self):
+    def invocation_details(self) -> None:
         return None
 
     @property
-    def outputs_dict(self):
+    def outputs_dict(self) -> Optional[Dict[str, Any]]:
         return self._outputs
 
 
-def run_cwltool(ctx, runnable, job_path, **kwds):
+def run_cwltool(
+    ctx: "PlanemoCliContext", runnable: "Runnable", job_path: str, **kwds
+) -> Union[ErrorRunResponse, CwlToolRunResponse]:
     """Translate planemo kwds to cwltool kwds and run cwltool main function."""
     ensure_cwltool_available()
 

planemo/cwl/toil.py

@@ -11,7 +11,10 @@
     error,
     real_io,
 )
-from planemo.runnable import ErrorRunResponse
+from planemo.runnable import (
+    ErrorRunResponse,
+    Runnable,
+)
 from .run import (
     CwlToolRunResponse,
     JSON_PARSE_ERROR_MESSAGE,
@@ -20,7 +23,7 @@
 TOIL_REQUIRED_MESSAGE = "This functionality requires Toil, please install with 'pip install toil'"
 
 
-def run_toil(ctx, path, job_path, **kwds):
+def run_toil(ctx, runnable: "Runnable", job_path: str, **kwds):
     """Translate planemo kwds to cwltool kwds and run cwltool main function."""
     _ensure_toil_available()
 
@@ -42,8 +45,8 @@ def run_toil(ctx, path, job_path, **kwds):
     if kwds.get("non_strict_cwl", False):
         args.append("--non-strict")
 
-    args.extend([path, job_path])
-    ctx.vlog("Calling cwltoil with arguments %s" % args)
+    args.extend([runnable.path, job_path])
+    ctx.vlog(f"Calling cwltoil with arguments {args}")
     with tempfile.NamedTemporaryFile("w") as tmp_stdout:
         # cwltool passes sys.stderr to subprocess.Popen - ensure it has
         # and actual fileno.
@@ -65,6 +68,7 @@ def run_toil(ctx, path, job_path, **kwds):
             return ErrorRunResponse("Error running Toil")
         outputs = result
     return CwlToolRunResponse(
+        runnable,
         "",
         outputs=outputs,
     )

planemo/engine/toil.py

@@ -17,8 +17,7 @@ def _run(self, runnables, job_paths):
         """Run CWL job using Toil."""
         results = []
         for runnable, job_path in zip(runnables, job_paths):
-            path = runnable.path
-            results.append(cwl.run_toil(self._ctx, path, job_path, **self._kwds))
+            results.append(cwl.run_toil(self._ctx, runnable, job_path, **self._kwds))
         return results
 
 

planemo/git.py

@@ -2,20 +2,29 @@
 
 import os
 import subprocess
+from typing import (
+    Dict,
+    List,
+    Optional,
+    TYPE_CHECKING,
+)
 
 from galaxy.util import unicodify
 
 from planemo import io
 
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
 
-def git_env_for(path):
+
+def git_env_for(path: str) -> Dict[str, str]:
     """Setup env dictionary to target specified git repo with git commands."""
     env = os.environ.copy()
     env.update({"GIT_WORK_DIR": path, "GIT_DIR": os.path.join(path, ".git")})
     return env
 
 
-def ls_remote(ctx, remote_repo):
+def ls_remote(ctx: "PlanemoCliContext", remote_repo: str) -> Dict[str, str]:
     """Return a dictionary with refs as key and commits as value."""
     commits_and_refs = io.communicate(
         ["git", "ls-remote", remote_repo],
@@ -24,22 +33,28 @@ def ls_remote(ctx, remote_repo):
     return dict(line.split()[::-1] for line in commits_and_refs.decode("utf-8").splitlines())
 
 
-def init(ctx, repo_path):
+def init(ctx: "PlanemoCliContext", repo_path: str) -> None:
     env = git_env_for(repo_path)
     io.communicate(["git", "init"], env=env)
 
 
-def add(ctx, repo_path, file_path):
+def add(ctx: "PlanemoCliContext", repo_path: str, file_path: str) -> None:
     env = git_env_for(repo_path)
     io.communicate(["git", "add", os.path.relpath(file_path, repo_path)], env=env, cwd=repo_path)
 
 
-def commit(ctx, repo_path, message=""):
+def commit(ctx: "PlanemoCliContext", repo_path: str, message: str = "") -> None:
     env = git_env_for(repo_path)
     io.communicate(["git", "commit", "-m", message], env=env)
 
 
-def push(ctx, repo_path, to=None, branch=None, force=False):
+def push(
+    ctx: "PlanemoCliContext",
+    repo_path: str,
+    to: Optional[str] = None,
+    branch: Optional[str] = None,
+    force: bool = False,
+) -> None:
     env = git_env_for(repo_path)
     cmd = ["git", "push"]
     if force:
@@ -71,7 +86,9 @@ def checkout(ctx, remote_repo, local_path, branch=None, remote="origin", from_br
         io.communicate(["git", "merge", "--ff-only", f"{remote}/{from_branch}"], env=env)
 
 
-def command_clone(ctx, src, dest, mirror=False, branch=None):
+def command_clone(
+    ctx: "PlanemoCliContext", src: str, dest: str, mirror: bool = False, branch: Optional[str] = None
+) -> List[str]:
     """Produce a command-line string to clone a repository.
 
     Take in ``ctx`` to allow more configurability down the road.
@@ -85,40 +102,38 @@ def command_clone(ctx, src, dest, mirror=False, branch=None):
     return cmd
 
 
-def diff(ctx, directory, range):
+def diff(ctx: "PlanemoCliContext", directory: str, range: str) -> List[str]:
     """Produce a list of diff-ed files for commit range."""
-    cmd_template = "cd '%s' && git diff --name-only '%s' --"
-    cmd = cmd_template % (directory, range)
+    cmd = f"cd '{directory}' && git diff --name-only '{range}' --"
     stdout, _ = io.communicate(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
     return [line.strip() for line in unicodify(stdout).splitlines() if line]
 
 
-def clone(*args, **kwds):
+def clone(*args, **kwds) -> None:
     """Clone a git repository.
 
     See :func:`command_clone` for description of arguments.
     """
     command = command_clone(*args, **kwds)
-    return io.communicate(command)
+    io.communicate(command)
 
 
-def rev(ctx, directory):
+def rev(ctx: "PlanemoCliContext", directory: str) -> str:
     """Raw revision for git directory specified.
 
     Throws ``RuntimeError`` if not a git directory.
     """
-    cmd_template = "cd '%s' && git rev-parse HEAD"
-    cmd = cmd_template % directory
+    cmd = f"cd '{directory}' && git rev-parse HEAD"
     stdout, _ = io.communicate(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     return unicodify(stdout).strip()
 
 
-def is_rev_dirty(ctx, directory):
+def is_rev_dirty(ctx: "PlanemoCliContext", directory: str) -> bool:
     """Check if specified git repository has uncommitted changes."""
     return io.shell(["git", "diff", "--quiet"], cwd=directory) != 0
 
 
-def rev_if_git(ctx, directory):
+def rev_if_git(ctx: "PlanemoCliContext", directory: str) -> Optional[str]:
     """Determine git revision (or ``None``)."""
     try:
         the_rev = rev(ctx, directory)

planemo/tools.py

@@ -3,22 +3,34 @@
 import os
 import sys
 import traceback
+from typing import (
+    Iterable,
+    Iterator,
+    List,
+    Tuple,
+    TYPE_CHECKING,
+    Union,
+)
 
 from galaxy.tool_util import loader_directory
 from galaxy.tool_util.fetcher import ToolLocationFetcher
+from galaxy.tool_util.loader_directory import is_tool_load_error
+from galaxy.tool_util.parser.interface import ToolSource
 
 from planemo.io import (
     error,
     info,
 )
 
-is_tool_load_error = loader_directory.is_tool_load_error
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
+
 SKIP_XML_MESSAGE = "Skipping XML file - does not appear to be a tool %s."
 SHED_FILES = ["tool_dependencies.xml", "repository_dependencies.xml"]
 LOAD_ERROR_MESSAGE = "Error loading tool with path %s"
 
 
-def uri_to_path(ctx, uri):
+def uri_to_path(ctx: "PlanemoCliContext", uri: str) -> str:
     """Fetch URI to a local path."""
     fetcher = ToolLocationFetcher()
     return fetcher.to_tool_path(uri)
@@ -34,7 +46,13 @@ def uris_to_paths(ctx, uris):
     return paths
 
 
-def yield_tool_sources_on_paths(ctx, paths, recursive=False, yield_load_errors=True, exclude_deprecated=False):
+def yield_tool_sources_on_paths(
+    ctx: "PlanemoCliContext",
+    paths: Iterable[str],
+    recursive: bool = False,
+    yield_load_errors: bool = True,
+    exclude_deprecated: bool = False,
+) -> Iterator[Tuple[str, Union[ToolSource, object]]]:
     """Walk paths and yield ToolSource objects discovered."""
     for path in paths:
         for (tool_path, tool_source) in yield_tool_sources(ctx, path, recursive, yield_load_errors):
@@ -43,7 +61,9 @@ def yield_tool_sources_on_paths(ctx, paths, recursive=False, yield_load_errors=T
             yield (tool_path, tool_source)
 
 
-def yield_tool_sources(ctx, path, recursive=False, yield_load_errors=True):
+def yield_tool_sources(
+    ctx: "PlanemoCliContext", path: str, recursive: bool = False, yield_load_errors: bool = True
+) -> Iterator[Tuple[str, Union[ToolSource, object]]]:
     """Walk single path and yield ToolSource objects discovered."""
     tools = load_tool_sources_from_path(
         path,
@@ -63,7 +83,9 @@ def yield_tool_sources(ctx, path, recursive=False, yield_load_errors=True):
         yield (tool_path, tool_source)
 
 
-def load_tool_sources_from_path(path, recursive, register_load_errors=False):
+def load_tool_sources_from_path(
+    path: str, recursive: bool, register_load_errors: bool = False
+) -> List[Tuple[str, Union[ToolSource, object]]]:
     """Generate a list for tool sources found down specified path."""
     return loader_directory.load_tool_sources_from_path(
         path,
@@ -78,7 +100,7 @@ def _load_exception_handler(path, exc_info):
     traceback.print_exception(*exc_info, limit=1, file=sys.stderr)
 
 
-def _is_tool_source(ctx, tool_path, tool_source):
+def _is_tool_source(ctx: "PlanemoCliContext", tool_path: str, tool_source: "ToolSource") -> bool:
     if os.path.basename(tool_path) in SHED_FILES:
         return False
     root = getattr(tool_source, "root", None)