Merge pull request #1320 from nsoranzo/type_annot_staging

Type annotation for input staging-related code

docs/planemo.rst

@@ -31,14 +31,6 @@ planemo.autoupdate module
    :undoc-members:
    :show-inheritance:
 
-planemo.bioblend module
------------------------
-
-.. automodule:: planemo.bioblend
-   :members:
-   :undoc-members:
-   :show-inheritance:
-
 planemo.ci module
 -----------------
 

planemo/commands/cmd_conda_build.py

@@ -1,20 +1,28 @@
 """Module describing the planemo ``conda_build`` command."""
+from typing import (
+    Tuple,
+    TYPE_CHECKING,
+)
 
 import click
 
 from planemo import options
 from planemo.cli import command_function
 from planemo.conda import build_conda_context
+from planemo.io import error
+
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
 
 
 @click.command("conda_build")
 @options.conda_target_options(include_local=False)  # No reason to expose local, we have to use it.
 @options.recipe_arg(multiple=True)
 @command_function
-def cli(ctx, paths, **kwds):
+def cli(ctx: "PlanemoCliContext", paths: Tuple[str], **kwds) -> None:
     """Perform conda build with Planemo's conda."""
     # Force conda_use_local for building...
     kwds["conda_use_local"] = True
     conda_context = build_conda_context(ctx, handle_auto_init=True, **kwds)
-    build_args = list(paths)
-    conda_context.exec_command("build", build_args)
+    if conda_context.exec_command("build", paths) != 0:
+        error(f"Failed to build [{' '.join(paths)}] with conda.")

planemo/commands/cmd_list_alias.py

@@ -23,7 +23,7 @@ def cli(ctx, profile, **kwds):
     """
     info("Looking for profiles...")
     aliases = profiles.list_alias(ctx, profile)
-    if tabulate:
+    if tabulate is not None:
         print(tabulate({"Alias": aliases.keys(), "Object": aliases.values()}, headers="keys"))
     else:
         print(json.dumps(aliases, indent=4, sort_keys=True))

planemo/commands/cmd_list_invocations.py

@@ -42,7 +42,7 @@ def cli(ctx, workflow_identifier, **kwds):
         key=profile["galaxy_admin_key"] or profile["galaxy_user_key"],
         workflow_id=workflow_id,
     )
-    if tabulate:
+    if tabulate is not None:
         state_colors = {
             "ok": "\033[92m",  # green
             "running": "\033[93m",  # yellow

planemo/commands/cmd_rerun.py

@@ -1,16 +1,24 @@
 """Module describing the planemo ``rerun`` command."""
+from typing import (
+    Tuple,
+    TYPE_CHECKING,
+)
 
 import click
 
 from planemo import options
 from planemo.cli import command_function
 from planemo.engine import engine_context
+from planemo.engine.galaxy import ExternalGalaxyEngine
 from planemo.io import (
     error,
     info,
 )
 from planemo.runnable import Rerunnable
 
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
+
 
 @click.command("rerun")
 @options.profile_option()
@@ -38,7 +46,7 @@
     nargs=-1,
 )
 @command_function
-def cli(ctx, rerunnable_ids, **kwds):
+def cli(ctx: "PlanemoCliContext", rerunnable_ids: Tuple[str], **kwds) -> None:
     """Planemo command for rerunning and remapping failed jobs on an external Galaxy server.
     Supply a list of history, invocation or job IDs, identifying the ID type using the
     --invocation, --history or --job flag, and all associated failed jobs will be rerun.
@@ -58,6 +66,7 @@ def cli(ctx, rerunnable_ids, **kwds):
     kwds["engine"] = "external_galaxy"
     rerun_successful = True
     with engine_context(ctx, **kwds) as engine:
+        assert isinstance(engine, ExternalGalaxyEngine)
         for rerunnable_id in rerunnable_ids:
             rerunnable = Rerunnable(rerunnable_id, kwds["rerunnable_type"], kwds["galaxy_url"])
             rerun_result = engine.rerun(ctx, rerunnable, **kwds)

planemo/commands/cmd_upload_data.py

@@ -39,7 +39,7 @@ def cli(ctx, runnable_identifier, job_path, new_job_path, **kwds):
     runnable = for_runnable_identifier(ctx, runnable_identifier, kwds)
     with engine_context(ctx, **kwds) as engine:
         with engine.ensure_runnables_served([runnable]) as config:
-            job, _, _ = stage_in(ctx, runnable, config, job_path, **kwds)
+            job, _ = stage_in(ctx, runnable, config, job_path, **kwds)
 
     rewrite_job_file(job_path, new_job_path, job)
     info(f"Files uploaded and new job file written to {new_job_path}")

planemo/conda.py

@@ -27,12 +27,6 @@
 if TYPE_CHECKING:
     from planemo.cli import PlanemoCliContext
 
-MESSAGE_ERROR_FAILED_INSTALL = "Attempted to install conda and failed."
-MESSAGE_ERROR_CANNOT_INSTALL = "Cannot install Conda - perhaps due to a failed installation or permission problems."
-MESSAGE_ERROR_NOT_INSTALLING = (
-    "Conda not configured - run ``planemo conda_init`` or pass ``--conda_auto_init`` to continue."
-)
-
 BEST_PRACTICE_CHANNELS = ["conda-forge", "bioconda", "defaults"]
 
 
@@ -61,19 +55,20 @@ def build_conda_context(ctx: "PlanemoCliContext", **kwds) -> CondaContext:
         failed = True
         if auto_init:
             if conda_context.can_install_conda():
-                if conda_util.install_conda(conda_context):
-                    error(MESSAGE_ERROR_FAILED_INSTALL)
+                if conda_util.install_conda(conda_context) != 0:
+                    error("Attempted to install conda and failed.")
                 else:
                     failed = False
             else:
-                error(MESSAGE_ERROR_CANNOT_INSTALL)
+                error("Cannot install Conda - perhaps due to a failed installation or permission problems.")
         else:
-            error(MESSAGE_ERROR_NOT_INSTALLING)
+            error("Conda not configured - run ``planemo conda_init`` or pass ``--conda_auto_init`` to continue.")
 
         if failed:
             raise ExitCodeException(EXIT_CODE_FAILED_DEPENDENCIES)
     if handle_auto_init:
-        conda_context.ensure_conda_build_installed_if_needed()
+        if conda_context.ensure_conda_build_installed_if_needed() != 0:
+            error("Attempted to install conda-build and failed.")
     return conda_context
 
 

planemo/engine/galaxy.py

@@ -2,24 +2,30 @@
 
 import abc
 import contextlib
+from typing import TYPE_CHECKING
 
 from galaxy.tool_util.verify import interactor
 
 from planemo import io
 from planemo.galaxy.activity import (
     execute,
     execute_rerun,
+    GalaxyBaseRunResponse,
 )
 from planemo.galaxy.config import external_galaxy_config
 from planemo.galaxy.serve import serve_daemon
 from planemo.runnable import (
     DelayedGalaxyToolTestCase,
     ExternalGalaxyToolTestCase,
     GALAXY_TOOLS_PREFIX,
+    Rerunnable,
     RunnableType,
 )
 from .interface import BaseEngine
 
+if TYPE_CHECKING:
+    from planemo.cli import PlanemoCliContext
+
 INSTALLING_MESSAGE = "Installing repositories - this may take some time..."
 
 
@@ -177,7 +183,7 @@ def ensure_runnables_served(self, runnables):
             config.install_workflows()
             yield config
 
-    def rerun(self, ctx, rerunnable, **kwds):
+    def rerun(self, ctx: "PlanemoCliContext", rerunnable: Rerunnable, **kwds) -> GalaxyBaseRunResponse:
         with self.ensure_runnables_served([]) as config:
             rerun_response = execute_rerun(ctx, config, rerunnable, **kwds)
             return rerun_response