feat: beautify show cli command with colorful structured output (cocoindex-io#404)

* feat: enhance flow string representation with structured output

* feat: improve show cli command with colorful structured output

* fix: improve error handling in string representation of Flow

* refactor: clean up flow format helper logic

* feat: implement color highlights using rich for flow formatting

* refactor: make reusable helpers in flow format method

Author: lemorage

pyproject.toml

@@ -9,7 +9,7 @@ description = "With CocoIndex, users declare the transformation, CocoIndex creat
 authors = [{ name = "CocoIndex", email = "[email protected]" }]
 readme = "README.md"
 requires-python = ">=3.11"
-dependencies = ["sentence-transformers>=3.3.1", "click>=8.1.8"]
+dependencies = ["sentence-transformers>=3.3.1", "click>=8.1.8", "rich>=14.0.0"]
 license = "Apache-2.0"
 urls = { Homepage = "https://cocoindex.io/" }
 

python/cocoindex/cli.py

@@ -1,6 +1,7 @@
 import asyncio
 import click
 import datetime
+from rich.console import Console
 
 from . import flow, lib
 from .setup import sync_setup, drop_setup, flow_names_with_setup, apply_setup_changes
@@ -52,11 +53,14 @@ def ls(show_all: bool):
 
 @cli.command()
 @click.argument("flow_name", type=str, required=False)
-def show(flow_name: str | None):
+@click.option("--color/--no-color", default=True)
+def show(flow_name: str | None, color: bool):
     """
-    Show the flow spec.
+    Show the flow spec in a readable format with colored output.
     """
-    click.echo(str(_flow_by_name(flow_name)))
+    flow = _flow_by_name(flow_name)
+    console = Console(no_color=not color)
+    console.print(flow._render_text())
 
 @cli.command()
 def setup():

python/cocoindex/flow.py

@@ -8,11 +8,14 @@
 import re
 import inspect
 import datetime
+import json
 
 from typing import Any, Callable, Sequence, TypeVar
 from threading import Lock
 from enum import Enum
 from dataclasses import dataclass
+from rich.text import Text
+from rich.console import Console
 
 from . import _engine
 from . import index
@@ -451,8 +454,58 @@ def _lazy_engine_flow() -> _engine.Flow:
             return engine_flow
         self._lazy_engine_flow = _lazy_engine_flow
 
+    def _format_flow(self, flow_dict: dict) -> Text:
+        output = Text()
+
+        def add_line(content, indent=0, style=None, end="\n"):
+            output.append(" " * indent)
+            output.append(content, style=style)
+            output.append(end)
+
+        def format_key_value(key, value, indent):
+            if isinstance(value, (dict, list)):
+                add_line(f"- {key}:", indent, style="green")
+                format_data(value, indent + 2)
+            else:
+                add_line(f"- {key}:", indent, style="green", end="")
+                add_line(f" {value}", style="yellow")
+
+        def format_data(data, indent=0):
+            if isinstance(data, dict):
+                for key, value in data.items():
+                    format_key_value(key, value, indent)
+            elif isinstance(data, list):
+                for i, item in enumerate(data):
+                    format_key_value(f"[{i}]", item, indent)
+            else:
+                add_line(str(data), indent, style="yellow")
+
+        # Header
+        flow_name = flow_dict.get("name", "Unnamed")
+        add_line(f"Flow: {flow_name}", style="bold cyan")
+
+        # Section
+        for section_title, section_key in [
+            ("Sources:", "import_ops"),
+            ("Processing:", "reactive_ops"),
+            ("Targets:", "export_ops"),
+        ]:
+            add_line("")
+            add_line(section_title, style="bold cyan")
+            format_data(flow_dict.get(section_key, []), indent=0)
+
+        return output
+
+    def _render_text(self) -> Text:
+        flow_spec_str = str(self._lazy_engine_flow())
+        try:
+            flow_dict = json.loads(flow_spec_str)
+            return self._format_flow(flow_dict)
+        except json.JSONDecodeError:
+            return Text(flow_spec_str)
+
     def __str__(self):
-        return str(self._lazy_engine_flow())
+        return str(self._render_text())
 
     def __repr__(self):
         return repr(self._lazy_engine_flow())