workflow progress bar.

Author: jmchilton

planemo/commands/cmd_workflow_test_on_invocation.py

@@ -15,9 +15,7 @@
 
 @click.command("workflow_test_on_invocation")
 @options.optional_tools_arg(multiple=False, allow_uris=False, metavar="TEST.YML")
-@options.required_invocation_id_arg()
-@options.galaxy_url_option(required=True)
-@options.galaxy_user_key_option(required=True)
+@options.invocation_target_options()
 @options.test_index_option()
 @options.test_output_options()
 @command_function

planemo/commands/cmd_workflow_track.py

@@ -0,0 +1,50 @@
+"""Module describing the planemo ``workflow_track`` command."""
+
+import click
+
+from planemo import options
+from planemo.cli import command_function
+from planemo.engine.factory import engine_context
+from planemo.galaxy.workflow_progress import WorkflowProgress
+
+
+@click.command("workflow_track")
+@options.invocation_target_options()
+@command_function
+def cli(ctx, invocation_id, **kwds):
+    """Run defined tests against existing workflow invocation."""
+    with WorkflowProgress() as workflow_progress:
+        workflow_progress.add_bars()
+        import time
+
+        time.sleep(1)
+        new_step = {"state": "new"}
+        scheduled_step = {"state": "scheduled"}
+        new_steps = [new_step, new_step, new_step]
+        one_scheduled_steps = [scheduled_step, new_step, new_step]
+        two_scheduled_steps = [scheduled_step, scheduled_step, new_step]
+        all_scheduled_steps = [scheduled_step, scheduled_step, scheduled_step]
+        state_pairs = [
+            ({"state": "new"}, {}),
+            ({"state": "ready", "steps": new_steps}, {}),
+            ({"state": "ready", "steps": one_scheduled_steps}, {"states": {"new": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 2}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 1, "running": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 1, "ok": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"ok": 2}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 2, "new": 3}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 2, "running": 1, "new": 2}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 3, "running": 1, "new": 1}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 4, "running": 1}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 5}}),
+        ]
+        for invocation, job_states_summary in state_pairs:
+            workflow_progress.handle_invocation(invocation, job_states_summary)
+            time.sleep(1)
+
+    with engine_context(ctx, engine="external_galaxy", **kwds) as engine, engine.ensure_runnables_served([]) as config:
+        user_gi = config.user_gi
+        invocation = user_gi.invocations.show_invocation(invocation_id)
+        # https://stackoverflow.com/questions/23113494/double-progress-bar-in-python
+
+    ctx.exit(0)

planemo/galaxy/activity.py

@@ -3,7 +3,6 @@
 import os
 import sys
 import tempfile
-import time
 import traceback
 from datetime import datetime
 from typing import (
@@ -42,6 +41,12 @@
     retry_on_timeouts,
     summarize_history,
 )
+from planemo.galaxy.invocations.api import BioblendInvocationApi
+from planemo.galaxy.invocations.polling import (
+    PollingTrackerImpl,
+)
+from planemo.galaxy.invocations.polling import wait_for_invocation_and_jobs as polling_wait_for_invocation_and_jobs
+from planemo.galaxy.invocations.progress import WorkflowProgressDisplay
 from planemo.io import wait_on
 from planemo.runnable import (
     ErrorRunResponse,
@@ -770,58 +775,15 @@ def _history_id(gi, **kwds) -> str:
 def wait_for_invocation_and_jobs(
     ctx, invocation_id: str, history_id: str, user_gi: GalaxyInstance, polling_backoff: int
 ):
-    ctx.vlog("Waiting for invocation [%s]" % invocation_id)
-    final_invocation_state = "new"
-
-    # TODO: hook in invocation["messages"]
-    error_message = ""
-    job_state = "ok"
-    try:
-        final_invocation_state = _wait_for_invocation(ctx, user_gi, invocation_id, polling_backoff)
-        assert final_invocation_state == "scheduled"
-    except Exception as e:
-        ctx.vlog(f"Problem waiting on invocation: {str(e)}")
-        summarize_history(ctx, user_gi, history_id)
-        error_message = f"Final state of invocation {invocation_id} is [{final_invocation_state}]"
-
-    ctx.vlog(f"Final state of invocation {invocation_id} is [{final_invocation_state}]")
-
-    job_state = _wait_for_invocation_jobs(ctx, user_gi, invocation_id, polling_backoff)
-    if job_state not in ("ok", "skipped"):
-        msg = f"Failed to run workflow, at least one job is in [{job_state}] state."
-        error_message = msg if not error_message else f"{error_message}. {msg}"
-    else:
-        # wait for possible subworkflow invocations
-        invocation = user_gi.invocations.show_invocation(invocation_id)
-        for step in invocation["steps"]:
-            if step.get("subworkflow_invocation_id") is not None:
-                final_invocation_state, job_state, error_message = wait_for_invocation_and_jobs(
-                    ctx,
-                    invocation_id=step["subworkflow_invocation_id"],
-                    history_id=history_id,
-                    user_gi=user_gi,
-                    polling_backoff=polling_backoff,
-                )
-                if final_invocation_state != "scheduled" or job_state not in ("ok", "skipped"):
-                    return final_invocation_state, job_state, error_message
-
-        ctx.vlog(f"The final state of all jobs and subworkflow invocations for invocation [{invocation_id}] is 'ok'")
-    return final_invocation_state, job_state, error_message
-
-
-def _wait_for_invocation(ctx, gi, invocation_id, polling_backoff=0):
-    def state_func():
-        return retry_on_timeouts(ctx, gi, lambda gi: gi.invocations.show_invocation(invocation_id))
-
-    return _wait_on_state(state_func, polling_backoff)
-
-
-def has_jobs_in_states(ctx, gi, history_id, states):
-    params = {"history_id": history_id}
-    jobs_url = gi.url + "/jobs"
-    jobs = gi.jobs._get(url=jobs_url, params=params)
-    target_jobs = [j for j in jobs if j["state"] in states]
-    return len(target_jobs) > 0
+    polling_tracker = PollingTrackerImpl(polling_backoff)
+    invocation_api = BioblendInvocationApi(ctx, user_gi)
+    with WorkflowProgressDisplay(invocation_id) as workflow_progress_display:
+        final_invocation_state, job_state, error_message = polling_wait_for_invocation_and_jobs(
+            ctx, invocation_id, invocation_api, polling_tracker, workflow_progress_display
+        )
+        if error_message:
+            summarize_history(ctx, user_gi, history_id)
+        return final_invocation_state, job_state, error_message
 
 
 def _wait_for_history(ctx, gi, history_id, polling_backoff=0):
@@ -835,19 +797,6 @@ def state_func():
     return _wait_on_state(state_func, polling_backoff)
 
 
-def _wait_for_invocation_jobs(ctx, gi, invocation_id, polling_backoff=0):
-    # Wait for invocation jobs to finish. Less brittle than waiting for a history to finish,
-    # as you could have more than one invocation in a history, or an invocation without
-    # steps that produce history items.
-
-    ctx.log(f"waiting for invocation {invocation_id}")
-
-    def state_func():
-        return retry_on_timeouts(ctx, gi, lambda gi: gi.jobs.get_jobs(invocation_id=invocation_id))
-
-    return _wait_on_state(state_func, polling_backoff)
-
-
 def _wait_for_job(gi, job_id, timeout=None):
     def state_func():
         return gi.jobs.show_job(job_id, full_details=True)

planemo/galaxy/invocations/__init__.py

@@ -0,0 +1,61 @@
+"""API interaction for Galaxy's workflow invocation API.
+
+Gives a mockable surface for testing, type contract consumed by Planemo,
+and builtin utilities around bioblend for working around transient request
+issues that have been observed in practice.
+"""
+
+from typing import (
+    Dict,
+    List,
+    Optional,
+    Protocol,
+)
+
+from typing_extensions import TypedDict
+
+from planemo.galaxy.api import retry_on_timeouts
+
+
+class InvocationStep(TypedDict, total=False):
+    state: Optional[str]
+    subworkflow_invocation_id: Optional[str]
+
+
+class Invocation(TypedDict, total=False):
+    id: str
+    state: str
+    steps: List[InvocationStep]
+
+
+class InvocationJobsSummary(TypedDict, total=False):
+    states: Dict[str, int]
+
+
+class InvocationApi(Protocol):
+
+    def get_invocation(self, invocation_id: str) -> Invocation: ...
+
+    def get_invocation_summary(self, invocation_id: str) -> InvocationJobsSummary: ...
+
+
+class BioblendInvocationApi(InvocationApi):
+
+    def __init__(self, ctx, user_gi):
+        self._ctx = ctx
+        self._user_gi = user_gi
+
+    def get_invocation(self, invocation_id: str) -> Invocation:
+        return retry_on_timeouts(self._ctx, self._user_gi, lambda gi: gi.invocations.show_invocation(invocation_id))
+
+    def get_invocation_summary(self, invocation_id: str) -> InvocationJobsSummary:
+        return retry_on_timeouts(
+            self._ctx, self._user_gi, lambda gi: gi.invocations.get_invocation_summary(invocation_id)
+        )
+
+
+def invocation_state_terminal(state: str):
+    return state in ["scheduled", "cancelled", "failed"]
+
+
+JOB_ERROR_STATES = ["error", "deleted", "failed", "stopped", "stop", "deleting"]