add list runs command

metaflow/plugins/__init__.py

@@ -84,6 +84,7 @@ def get_plugin_cli():
     from .aws.step_functions import step_functions_cli
     from .argo import argo_workflows_cli
     from .cards import card_cli
+    from . import list_cli
 
     return _ext_plugins["get_plugin_cli"]() + [
         package_cli.cli,
@@ -92,6 +93,7 @@ def get_plugin_cli():
         kubernetes_cli.cli,
         step_functions_cli.cli,
         argo_workflows_cli.cli,
+        list_cli.cli,
     ]
 
 

metaflow/plugins/list_cli.py

@@ -0,0 +1,105 @@
+from metaflow._vendor import click
+from metaflow import Flow, namespace
+from metaflow import util
+from metaflow.exception import MetaflowNotFound, CommandException
+import json
+
+
+@click.group()
+def cli():
+    pass
+
+
+@cli.group(help="List objects pertaining to your flow.")
+def list():
+    pass
+
+
+def _fetch_runs(flow_name, num_runs):
+    counter = 1
+    try:
+        flow = Flow(flow_name)
+    except MetaflowNotFound:
+        flow = None
+    run_list = []
+    if flow:
+        for run in Flow(flow_name).runs():
+            if counter > num_runs:
+                break
+            counter += 1
+            run_list.append(
+                dict(
+                    created=str(run.created_at),
+                    name=flow_name,
+                    id=run.id,
+                    status=run.successful,
+                    finished=run.finished,
+                    tags=[t for t in run.tags],
+                )
+            )
+    return run_list
+
+
+@click.option(
+    "--num-runs",
+    default=10,
+    type=click.IntRange(1, None),
+    help="Number of runs to show.",
+)
+@click.option(
+    "--as-json",
+    default=False,
+    is_flag=True,
+    help="Print run list as a JSON object",
+)
+@click.option(
+    "--file",
+    default=None,
+    help="Save the run list to file as json.",
+)
+@click.option("--user", default=None, help="List runs for the given user.")
+@click.option(
+    "--namespace", "ns", default=None, help="List runs only for the given namespace."
+)
+@click.option(
+    "--all",
+    default=False,
+    is_flag=True,
+    help="List runs from the global namespace instead of the current user.",
+)
+@list.command(help="List recent runs for your flow.")
+@click.pass_context
+def runs(ctx, num_runs, as_json, file, user, all, ns):
+    if user and all:
+        raise CommandException("--user and --all are mutually exclusive.")
+    if user and ns:
+        raise CommandException("--user and --namespace are mutually exclusive.")
+    if all and ns:
+        raise CommandException("--all and --namespace are mutually exclusive.")
+
+    if all:
+        namespace(None)
+    elif user:
+        namespace("user:{}".format(user))
+    elif ns:
+        namespace(ns)
+    else:
+        namespace("user:{}".format(util.get_username()))
+
+    run_list = _fetch_runs(ctx.obj.flow.name, num_runs)
+    if not run_list:
+        ctx.obj.echo("No runs found for flow: {name}".format(name=ctx.obj.flow.name))
+        return
+
+    if file:
+        with open(file, "w") as f:
+            json.dump(run_list, f)
+    if as_json:
+        ctx.obj.echo(json.dumps(run_list, indent=4), err=False)
+    else:
+        for run in run_list:
+            ctx.obj.echo(
+                "{created} {name} [{id}] (Successful:{status} Finished:{finished})".format(
+                    **run
+                )
+            )

test/core/metaflow_test/cli_check.py

@@ -167,6 +167,13 @@ def _list_cards(self, step, task=None, card_type=None):
             with open(f.name, "r") as jsf:
                 return json.load(jsf)
 
+    def list_runs(self):
+        with NamedTemporaryFile(dir=".") as f:
+            cmd = ["--quiet", "list", "runs", "--file", f.name]
+            self.run_cli(cmd)
+            with open(f.name, "r") as f:
+                return json.load(f)
+
     def get_card(self, step, task, card_type, card_hash=None, card_id=None):
         with NamedTemporaryFile(dir=".") as f:
             cmd = [

test/core/tests/list_runs.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+from metaflow_test import MetaflowTest, assert_equals, steps
+
+
+class ListRunsTest(MetaflowTest):
+    """
+    Test that tags are assigned properly.
+    """
+
+    PRIORITY = 1
+
+    @steps(1, ["all"])
+    def step_all(self):
+        self.flow_run_id = current.run_id
+
+    def check_results(self, flow, checker):
+        if type(checker).__name__ == "CliCheck":
+            from metaflow import Flow
+
+            cli_run_list = checker.list_runs()
+            assert cli_run_list, "No Runs Returned in CLI for {name}".format(
+                name=flow.name
+            )
+
+            latest_run_id = Flow(flow.name).latest_run.data.flow_run_id
+            latest_run_found = False
+            for run in cli_run_list:
+                assert_equals(run["name"], flow.name)
+                if run["id"] == latest_run_id:
+                    latest_run_found = True
+
+            assert (
+                latest_run_found
+            ), "Latest run {id} for {name} not listed in cli.".format(
+                id=latest_run_id, name=flow.name
+            )