this update will make it possible to list all invocations without knowing the workflow id, invocations will then be grouped by workflow. The user will also have to option of outputing the list as a json.

Author: Smeds

planemo/commands/cmd_list_invocations.py

@@ -7,7 +7,10 @@
 from planemo import options
 from planemo.cli import command_function
 from planemo.galaxy import profiles
-from planemo.galaxy.api import get_invocations
+from planemo.galaxy.api import (
+    get_invocations,
+    gi,
+)
 from planemo.galaxy.workflows import remote_runnable_to_workflow_id
 from planemo.io import (
     error,
@@ -25,59 +28,106 @@
 @click.argument(
     "workflow_identifier",
     type=click.STRING,
+    required=False,
+    default="",
+)
+@click.option(
+    "--raw",
+    is_flag=True,
+    help="output will be a json structure.",
+    default=False,
+)
+@click.option(
+    "--max_items",
+    type=click.INT,
+    help="max number of items returned.",
+    default=100,
+)
+@click.option(
+    "--offset_items",
+    type=click.INT,
+    help="skip first X items.",
+    default=0,
 )
 @options.profile_option(required=True)
 @command_function
-def cli(ctx, workflow_identifier, **kwds):
+def cli(ctx, workflow_identifier, raw, max_items, offset_items, **kwds):
     """
     Get a list of invocations for a particular workflow ID or alias.
     """
-    info(f"Looking for invocations for workflow {workflow_identifier}...")
-    runnable = for_runnable_identifier(ctx, workflow_identifier, kwds)
+    if not raw:
+        info(f"Looking for invocations for workflow {workflow_identifier}...")
     profile = profiles.ensure_profile(ctx, kwds.get("profile"))
-    assert runnable.is_remote_workflow_uri
-    workflow_id = remote_runnable_to_workflow_id(runnable)
-
+    if workflow_identifier:
+        runnable = for_runnable_identifier(ctx, workflow_identifier, kwds)        
+        assert runnable.is_remote_workflow_uri
+        workflow_id = remote_runnable_to_workflow_id(runnable)
+    else:
+        workflow_id = ""
+    gi_client = gi(None, profile["galaxy_url"], profile["galaxy_admin_key"] or profile["galaxy_user_key"])
     invocations = get_invocations(
-        url=profile["galaxy_url"],
-        key=profile["galaxy_admin_key"] or profile["galaxy_user_key"],
+        gi=gi_client,
         workflow_id=workflow_id,
+        instance=True,
+        max_items=max_items,
+        offset_items=offset_items,
     )
+    if raw:
+        print(json.dumps(invocations, indent=4, sort_keys=True))
+        return
     if tabulate is not None:
         state_colors = {
             "ok": "\033[92m",  # green
             "running": "\033[93m",  # yellow
             "error": "\033[91m",  # red
             "paused": "\033[96m",  # cyan
             "deleted": "\033[95m",  # magenta
+            "deleting": "\033[95m",  # magenta
             "deleted_new": "\033[95m",  # magenta
             "new": "\033[96m",  # cyan
             "queued": "\033[93m",  # yellow
+            "skipped": "\033[90m",  # gray
         }
-        print(
-            tabulate(
-                {
-                    "Invocation ID": invocations.keys(),
-                    "Jobs status": [
-                        ", ".join([f"{state_colors[k]}{v} jobs {k}\033[0m" for k, v in inv["states"].items()])
-                        for inv in invocations.values()
-                    ],
-                    "Invocation report URL": [
-                        "{}/workflows/invocations/report?id={}".format(profile["galaxy_url"].strip("/"), inv_id)
-                        for inv_id in invocations
-                    ],
-                    "History URL": [
-                        "{}/histories/view?id={}".format(
-                            profile["galaxy_url"].strip("/"), invocations[inv_id]["history_id"]
-                        )
-                        for inv_id in invocations
-                    ],
-                },
-                headers="keys",
+        def generate_column_data(data):
+            return {
+                "Invocation ID": data.keys(),
+                "Invocation report URL": [
+                    "{}/workflows/invocations/report?id={}".format(profile["galaxy_url"].strip("/"), inv_id)
+                    for inv_id in data
+                ],
+                "History URL": [
+                    "{}/histories/view?id={}".format(
+                        profile["galaxy_url"].strip("/"), invocations[inv_id]["history_id"]
+                    )
+                    for inv_id in data
+                ],
+                "Jobs status": [
+                    ", ".join([f"{state_colors[k]}{v} jobs {k}\033[0m" for k, v in inv["states"].items()])
+                    for inv in data.values()
+                ],
+            }
+
+        grouped_invocations = {}
+        workflows = {}
+        for inv_id, inv in invocations.items():
+            wf_id = inv["workflow_id"]
+            if wf_id not in grouped_invocations:
+                workflow = gi_client.workflows.show_workflow(workflow_id=wf_id, instance=True)
+                workflows[wf_id] = (workflow["name"], workflow["id"])
+            grouped_invocations.setdefault(wf_id, {})[inv_id] = inv
+        for workflow_id in grouped_invocations:
+            header = f"Workflow: {workflows[workflow_id][0]} : {profile['galaxy_url'].strip('/')}/workflows/run?id={workflows[workflow_id][1]}"
+            print(f"\n{header}")
+            print(len(header) * "=")
+            print(
+                tabulate(
+                    generate_column_data((grouped_invocations[workflow_id])),
+                    headers="keys",
+                )
             )
-        )
     else:
         error("The tabulate package is not installed, invocations could not be listed correctly.")
         print(json.dumps(invocations, indent=4, sort_keys=True))
-    info(f"{len(invocations)} invocations found.")
+    if not raw:
+        info(f"{len(invocations)} invocations found.")
     return

planemo/galaxy/api.py

@@ -114,12 +114,21 @@ def summarize_history(ctx, gi, history_id):
         print("|")
 
 
-def get_invocations(url, key, workflow_id):
-    inv_gi = gi(None, url, key)
-    invocations = inv_gi.workflows.get_invocations(workflow_id)
+def get_invocations(gi, workflow_id, instance=False, max_items=100, items_per_request=20, offset_items=0):
+    invocations = []
+    while len(invocations) < max_items:
+        if workflow_id:
+            items = gi.invocations.get_invocations(workflow_id,limit=min(items_per_request, max_items), offset=len(invocations)+offset_items)
+        else:
+            items = gi.invocations.get_invocations(instance=instance, limit=min(items_per_request, max_items), offset=len(invocations)+offset_items)
+        if (items is None) or (len(items) == 0):
+            break
+        else:
+            invocations.extend(items)
     return {
         invocation["id"]: {
-            "states": inv_gi.invocations.get_invocation_summary(invocation["id"])["states"],
+            "states": gi.invocations.get_invocation_summary(invocation["id"])["states"],
+            "workflow_id": invocation["workflow_id"],
             "history_id": invocation["history_id"],
         }
         for invocation in invocations

tests/test_run.py

@@ -52,6 +52,7 @@ def test_run_cat_cwltool_more_options(self):
                 job_path,
             ]
             self._check_exit_code(test_cmd)
+            
             assert os.path.exists(os.path.join(f, "tool_test_output.html"))
             assert os.path.exists(os.path.join(f, "tool_test_output.json"))
 
@@ -97,7 +98,7 @@ def test_run_cat(self):
     @skip_if_environ("PLANEMO_SKIP_GALAXY_CWL_TESTS")
     def test_run_output_directory(self):
         with self._isolate() as f:
-            tool_path = _cwl_file("wc-tool.cwl")
+            tool_path = _cwl_file("wc-tool1.0.cwl")
             job_path = _cwl_file("wc-job.json")
             test_cmd = [
                 "--verbose",
@@ -115,3 +116,34 @@ def test_run_output_directory(self):
             assert os.path.exists(output_path)
             with open(output_path) as fh:
                 assert fh.read().startswith("  16  198 1111")
+
+    @skip_if_environ("PLANEMO_SKIP_GALAXY_TESTS")
+    @skip_if_environ("PLANEMO_SKIP_CWLTOOL_TESTS")
+    @skip_if_environ("PLANEMO_SKIP_GALAXY_CWL_TESTS")
+    def test_run_download_output(self):
+
+        with self._isolate() as f:
+            tool_path = _cwl_file("wc-tool1.0.cwl")
+            job_path = _cwl_file("wc-job.json")
+            test_cmd = [
+                "--verbose",
+                "run",
+                "--no_dependency_resolution",
+                "--output_directory",
+                f,
+                "--download_outputs",
+                tool_path,
+                job_path,
+            ]
+            self._check_exit_code(test_cmd)
+            assert os.path.exists(os.path.join(f, "tool_test_output.html"))
+            assert os.path.exists(os.path.join(f, "tool_test_output.json"))
+            output_path = os.path.join(f, "output")
+            assert os.path.exists(output_path)
+            files = os.listdir(f)
+            print(files)
+            assert False, files
+            assert "tool_test_output.html" in files
+            assert "tool_test_output.json" in files
+            with open(output_path) as fh:
+                assert fh.read().startswith("  16  198 1111")