feat(cli/results.py): adding results cli (#23)

workflow/cli/main.py

@@ -4,6 +4,7 @@
 
 # from workflow.cli.buckets import buckets
 from workflow.cli.pipelines import pipelines
+from workflow.cli.results import results
 from workflow.cli.run import run
 from workflow.cli.workspace import workspace
 
@@ -16,6 +17,7 @@ def cli():
 
 cli.add_command(run)
 # cli.add_command(buckets)
+cli.add_command(results)
 cli.add_command(pipelines)
 cli.add_command(workspace)
 

workflow/cli/results.py

@@ -0,0 +1,153 @@
+"""Results CLI Interface."""
+
+from json import dumps
+from typing import Optional
+
+import click
+from rich import pretty
+from rich.console import Console
+from rich.json import JSON
+from rich.table import Table
+from rich.text import Text
+
+from workflow.http.context import HTTPContext
+
+pretty.install()
+console = Console()
+
+
+table = Table(
+    title="\nWorkflow Results",
+    show_header=True,
+    header_style="magenta",
+    title_style="bold magenta",
+)
+status_colors = {
+    "success": "green",
+    "failure": "red",
+}
+yes_no_colors = {"yes": "green", "no": "red"}
+
+
+@click.group(name="results", help="Manage workflow results.")
+def results():
+    """Manage workflow results."""
+    pass
+
+
+@results.command("version", help="Show the version.")
+def version():
+    """Show the version."""
+    http = HTTPContext()
+    console.print(http.results.info())
+
+
+@results.command("count", help="Returns count of all pipelines on results backend.")
+@click.option("--status", "-s", help="Count by status")
+def count(status: Optional[str] = None):
+    """Count pipelines on results backend."""
+    http = HTTPContext()
+    count_result = http.results.count()
+    table.add_column("Pipeline", max_width=50, justify="left", style="bright_blue")
+    table.add_column("Count", max_width=50, justify="left")
+    for pipeline, count in count_result.items():
+        table.add_row(pipeline, str(count))
+    console.print(table)
+
+
+@results.command("view", help="Returns Results from the specified pipeline.")
+@click.argument("pipeline", type=str, required=True)
+@click.option("status", "-s", type=str, required=False, help="Filter by status.")
+@click.option("skip", "-k", type=int, required=False, help="Skip n results.")
+@click.option(
+    "limit", "-l", type=int, default=50, required=False, help="Deliver only n results."
+)
+def view(
+    pipeline: str,
+    status: Optional[str] = None,
+    skip: Optional[int] = None,
+    limit: Optional[int] = None,
+):
+    """View a set of filtered Results.
+
+    Parameters
+    ----------
+    pipeline : str
+        Pipeline name.
+    status : Optional[str], optional
+        Results status, by default None
+    skip : Optional[int], optional
+        Skip n results, by default None
+    limit : Optional[int], optional
+        Deliver n results, by default None
+    """
+    http = HTTPContext()
+    payload = {
+        "query": {"pipeline": pipeline},
+        "projection": {
+            "id": True,
+            "pipeline": True,
+            "products": True,
+            "plots": True,
+            "function": True,
+            "command": True,
+            "status": True,
+        },
+    }
+    if status:
+        payload["query"].update({"status": status})  # type: ignore
+    if skip:
+        payload.update({"skip": skip})  # type: ignore
+    if limit:
+        payload.update({"limit": limit})  # type: ignore
+
+    view_result = http.results.view(payload)
+    table.add_column("ID", max_width=50, justify="left", style="bright_blue")
+    table.add_column("Pipeline", max_width=50, justify="left", style="bright_blue")
+    table.add_column("Products", max_width=50, justify="left")
+    table.add_column("Plots", max_width=50, justify="left")
+    table.add_column("Function", max_width=50, justify="left")
+    table.add_column("Command", max_width=50, justify="left")
+    table.add_column("Status", max_width=50, justify="left")
+    for result in view_result:
+        result_status = Text(result["status"], style=status_colors[result["status"]])
+        has_products = "Yes" if result["products"] else "No"
+        has_products = Text(has_products, style=yes_no_colors[has_products.lower()])
+        has_plots = "Yes" if result["plots"] else "No"
+        has_plots = Text(has_plots, style=yes_no_colors[has_plots.lower()])
+        _function = (
+            result["function"] if result["function"] else Text("-------", style="red")
+        )
+        table.add_row(
+            result["id"],
+            result["pipeline"],
+            has_products,
+            has_plots,
+            _function,
+            " ".join(result["command"]),
+            result_status,
+        )
+    console.print(table)
+
+
+@results.command("inspect", help="Returns detailed information about specified Result.")
+@click.argument("pipeline", type=str, required=True)
+@click.argument("id", type=str, required=True)
+def inspect(pipeline: str, id: str):
+    """Returns details for one specific results.
+
+    Parameters
+    ----------
+    pipeline : str
+        Pipeline name.
+    id : str
+        Results ID.
+    """
+    http = HTTPContext()
+    payload = {
+        "query": {"pipeline": pipeline, "id": id},
+    }
+    inspect_result = http.results.view(payload)[0]
+    table.add_column(f"{pipeline} - {id}", max_width=100, justify="left")
+    table.add_row(JSON(dumps(inspect_result), indent=2))
+    console.print(table)

workflow/http/results.py

@@ -1,6 +1,11 @@
 """Workflow Results API."""
 
+from typing import Any, Dict, List
+
+from requests.models import Response
+
 from workflow.http.client import Client
+from workflow.utils.decorators import try_request
 
 
 class Results(Client):
@@ -13,4 +18,49 @@ class Results(Client):
         Results: A client for interacting with the Buckets backend.
     """
 
-    pass
+    @try_request
+    def info(self) -> Dict[str, Any]:
+        """Get the version of the results backend.
+
+        Returns:
+            Dict[str, Any]: The version of the results backend.
+        """
+        client_info = self.model_dump()
+        with self.session as session:
+            response: Response = session.get(url=f"{self.baseurl}/version")
+            response.raise_for_status()
+        server_info = response.json()
+        return {"client": client_info, "server": server_info}
+
+    @try_request
+    def count(self) -> Dict[str, Any]:
+        """Gets count of pipelines on results backend.
+
+        Returns
+        -------
+        Dict[str, Any]
+            Count payload.
+        """
+        with self.session as session:
+            response: Response = session.get(url=f"{self.baseurl}/status")
+            response.raise_for_status()
+        return response.json()
+
+    @try_request
+    def view(self, payload: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """Queries details for a specific Result.
+
+        Parameters
+        ----------
+        payload : Dict[str, Any]
+            Query payload.
+
+        Returns
+        -------
+        List[Dict[str, Any]]
+            Filtered results.
+        """
+        with self.session as session:
+            response: Response = session.post(url=f"{self.baseurl}/view", json=payload)
+            response.raise_for_status()
+        return response.json()