feat: enhance CLI usability with improved shell

.gitignore

@@ -129,3 +129,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Git worktrees
+.worktrees/

milvus_cli/CliClient.py

@@ -7,6 +7,7 @@
 from AliasClient import MilvusClientAlias
 from PartitionClient import MilvusClientPartition
 from RoleClient import MilvusClientRole
+from OutputFormatter import OutputFormatter
 from pymilvus import __version__
 
 
@@ -38,6 +39,7 @@ def __init__(self):
         self.role = MilvusClientRole(self.connection)
         self.alias = MilvusClientAlias(self.connection)
         self.partition = MilvusClientPartition(self.connection)
+        self.formatter = OutputFormatter()
 
     def connect(self, uri=None, token=None, tlsmode=0, cert=None):
         """

milvus_cli/OutputFormatter.py

@@ -0,0 +1,136 @@
+import json
+import csv
+import io
+from tabulate import tabulate
+
+
+class OutputFormatter:
+    """Global output formatter supporting table, json, and csv formats."""
+
+    FORMATS = ["table", "json", "csv"]
+    DEFAULT_TABLEFMT = "grid"
+
+    def __init__(self):
+        self._format = "table"
+
+    @property
+    def format(self):
+        return self._format
+
+    @format.setter
+    def format(self, value):
+        if value not in self.FORMATS:
+            raise ValueError(f"Invalid format '{value}'. Must be one of: {', '.join(self.FORMATS)}")
+        self._format = value
+
+    def format_output(self, data, headers=None, tablefmt="grid"):
+        """
+        Format data according to current format setting.
+
+        Args:
+            data: List of dicts, list of lists, or single dict
+            headers: Optional list of header names (for list of lists)
+            tablefmt: Table format for tabulate (default: grid)
+
+        Returns:
+            Formatted string
+        """
+        if not data:
+            return "No data to display."
+
+        # Type validation
+        if not isinstance(data, (dict, list)):
+            raise ValueError(f"Invalid data type '{type(data).__name__}'. Must be dict, list of dicts, or list of lists.")
+
+        # Normalize data to list of dicts
+        if isinstance(data, dict):
+            data = [data]
+
+        # Validate list contents
+        if data and not isinstance(data[0], (dict, list, tuple)):
+            raise ValueError(f"Invalid data type. List elements must be dicts or lists, got '{type(data[0]).__name__}'.")
+
+        if self._format == "json":
+            return self._format_json(data)
+        elif self._format == "csv":
+            return self._format_csv(data, headers)
+        else:  # table
+            return self._format_table(data, headers, tablefmt)
+
+    def _format_json(self, data):
+        """Format data as JSON."""
+        return json.dumps(data, indent=2, default=str, ensure_ascii=False)
+
+    def _format_csv(self, data, headers=None):
+        """Format data as CSV."""
+        if not data:
+            return ""
+
+        output = io.StringIO()
+
+        if isinstance(data[0], dict):
+            headers = headers or list(data[0].keys())
+            writer = csv.DictWriter(output, fieldnames=headers)
+            writer.writeheader()
+            writer.writerows(data)
+        else:
+            writer = csv.writer(output)
+            if headers:
+                writer.writerow(headers)
+            writer.writerows(data)
+
+        return output.getvalue().strip()
+
+    def _format_table(self, data, headers=None, tablefmt="grid"):
+        """Format data as table using tabulate."""
+        if not data:
+            return "No data to display."
+
+        if isinstance(data[0], dict):
+            headers = headers or list(data[0].keys())
+            rows = [list(row.values()) for row in data]
+        else:
+            rows = data
+
+        return tabulate(rows, headers=headers, tablefmt=tablefmt)
+
+    def format_list(self, items, header="Item"):
+        """Format a simple list of items."""
+        if not items:
+            return "No items to display."
+
+        if self._format == "json":
+            return json.dumps(items, indent=2, ensure_ascii=False)
+        elif self._format == "csv":
+            output = io.StringIO()
+            writer = csv.writer(output)
+            writer.writerow([header])
+            for item in items:
+                writer.writerow([item])
+            return output.getvalue().strip()
+        else:
+            data = [[item] for item in items]
+            return tabulate(data, headers=[header], tablefmt=self.DEFAULT_TABLEFMT)
+
+    def format_key_value(self, data, key_header="Property", value_header="Value"):
+        """Format key-value pairs (for show commands)."""
+        if not data:
+            return "No data to display."
+
+        if isinstance(data, dict):
+            items = list(data.items())
+        else:
+            items = data  # Assume list of [key, value] pairs
+
+        if self._format == "json":
+            if isinstance(data, dict):
+                return json.dumps(data, indent=2, default=str, ensure_ascii=False)
+            return json.dumps(dict(items), indent=2, default=str, ensure_ascii=False)
+        elif self._format == "csv":
+            output = io.StringIO()
+            writer = csv.writer(output)
+            writer.writerow([key_header, value_header])
+            writer.writerows(items)
+            return output.getvalue().strip()
+        else:
+            return tabulate(items, headers=[key_header, value_header], tablefmt="grid")

milvus_cli/history.py

@@ -0,0 +1,79 @@
+import json
+import os
+from datetime import datetime
+from pathlib import Path
+
+
+class ConnectionHistory:
+    """Manages persistent connection history stored in ~/.milvus_cli_connections.json"""
+
+    DEFAULT_PATH = Path.home() / ".milvus_cli_connections.json"
+
+    def __init__(self, path=None):
+        self.path = Path(path) if path else self.DEFAULT_PATH
+        self._connections = self._load()
+
+    def _load(self):
+        """Load connections from JSON file."""
+        if not self.path.exists():
+            return {}
+        try:
+            with open(self.path, "r") as f:
+                return json.load(f)
+        except (json.JSONDecodeError, IOError):
+            return {}
+
+    def _save(self):
+        """Save connections to JSON file."""
+        import click
+        try:
+            with open(self.path, "w") as f:
+                json.dump(self._connections, f, indent=2)
+        except IOError as e:
+            click.echo(f"Warning: Could not save connection history: {e}", err=True)
+
+    def save_connection(self, uri, token=None, tlsmode=0, cert=None, alias=None):
+        """
+        Save a connection. Uses URI as alias if not provided.
+        Updates timestamp if URI already exists.
+        """
+        key = alias if alias else uri
+        self._connections[key] = {
+            "uri": uri,
+            "token": token,
+            "tlsmode": tlsmode,
+            "cert": cert,
+            "last_used": datetime.now().isoformat(),
+        }
+        self._save()
+
+    def get_connection(self, alias):
+        """Get connection by alias."""
+        return self._connections.get(alias)
+
+    def list_connections(self):
+        """Return list of saved connections with their details."""
+        result = []
+        for alias, conn in self._connections.items():
+            result.append({
+                "alias": alias,
+                "uri": conn["uri"],
+                "last_used": conn.get("last_used", "unknown"),
+            })
+        return sorted(result, key=lambda x: x.get("last_used", ""), reverse=True)
+
+    def delete_connection(self, uri):
+        """Delete connection by URI. Returns True if found and deleted."""
+        # Find and remove any connection with matching URI
+        to_delete = [k for k, v in self._connections.items() if v["uri"] == uri]
+        if not to_delete:
+            return False
+        for key in to_delete:
+            del self._connections[key]
+        self._save()
+        return True
+
+    def clear(self):
+        """Clear all saved connections."""
+        self._connections = {}
+        self._save()

milvus_cli/prompt_style.py

@@ -0,0 +1,134 @@
+from prompt_toolkit.lexers import Lexer
+from prompt_toolkit.completion import Completer as PTCompleter, Completion
+from prompt_toolkit.styles import Style
+
+
+# Milvus CLI commands for syntax highlighting
+COMMANDS = {
+    "connect", "exit", "help", "clear", "version", "server_version",
+    "show", "set", "list", "create", "delete", "rename", "load", "release",
+    "use", "search", "query", "insert", "upsert", "grant", "revoke",
+    "flush", "compact", "truncate", "bulk_insert", "history", "get",
+    "describe", "import", "wait_for_loading",
+}
+
+SUBCOMMANDS = {
+    "collection", "database", "partition", "index", "user", "role", "alias",
+    "connections", "collections", "databases", "partitions", "indexes",
+    "users", "roles", "grants", "aliases", "output", "file", "row",
+    "connection_history", "bulk_insert_tasks", "bulk_insert_state",
+    "loading_progress", "index_progress", "query_segment_info",
+    "compaction_state", "compaction_plans", "replicas", "load_state",
+}
+
+OPTIONS = {
+    "-uri", "--uri", "-t", "--token", "-tls", "--tlsmode", "-cert", "--cert",
+    "-c", "--collection", "-p", "--partition", "-db", "--db_name",
+    "-f", "--fields", "-q", "--query", "-o", "--output", "--save-as",
+}
+
+
+class MilvusLexer(Lexer):
+    """Syntax highlighter for Milvus CLI commands."""
+
+    def lex_document(self, document):
+        def get_tokens(line_number):
+            line = document.lines[line_number]
+            tokens = []
+            pos = 0
+
+            for word in line.split():
+                # Find word position
+                start = line.find(word, pos)
+
+                # Add whitespace before word
+                if start > pos:
+                    tokens.append(("", line[pos:start]))
+
+                # Determine token style
+                if word in COMMANDS:
+                    style = "class:command"
+                elif word in SUBCOMMANDS:
+                    style = "class:subcommand"
+                elif word in OPTIONS or word.startswith("-"):
+                    style = "class:option"
+                elif word.startswith(("http://", "https://")):
+                    style = "class:uri"
+                else:
+                    style = ""
+
+                tokens.append((style, word))
+                pos = start + len(word)
+
+            # Add trailing whitespace
+            if pos < len(line):
+                tokens.append(("", line[pos:]))
+
+            return tokens
+
+        return get_tokens
+
+
+class MilvusCompleter(PTCompleter):
+    """
+    prompt_toolkit completer that wraps the existing Completer.
+    """
+
+    def __init__(self, old_completer):
+        self.old_completer = old_completer
+
+    def get_completions(self, document, complete_event):
+        text = document.text_before_cursor
+        words = text.split()
+
+        # Get the word being typed
+        word_before_cursor = document.get_word_before_cursor()
+
+        if not words:
+            # Show all commands
+            for cmd in sorted(self.old_completer.COMMANDS):
+                yield Completion(cmd, start_position=0, display_meta="command")
+            return
+
+        # If text ends with space, we're completing a new word
+        if text.endswith(" "):
+            words.append("")
+
+        cmd = words[0]
+
+        if len(words) == 1:
+            # Complete command name
+            for c in sorted(self.old_completer.COMMANDS):
+                if c.startswith(cmd):
+                    yield Completion(
+                        c,
+                        start_position=-len(cmd),
+                        display_meta="command"
+                    )
+        else:
+            # Complete subcommand or argument
+            if cmd in self.old_completer.COMMANDS:
+                impl = getattr(self.old_completer, f"complete_{cmd}", None)
+                if impl:
+                    args = words[1:]
+                    completions = impl(args)
+                    if completions:
+                        current = args[-1] if args else ""
+                        for comp in completions:
+                            comp = comp.rstrip()
+                            if comp and (not current or comp.startswith(current)):
+                                yield Completion(
+                                    comp,
+                                    start_position=-len(current),
+                                    display_meta="subcommand"
+                                )
+
+
+# Style for syntax highlighting
+milvus_style = Style.from_dict({
+    "command": "#00aa00 bold",      # Green bold for commands
+    "subcommand": "#0088ff",         # Blue for subcommands
+    "option": "#ffaa00",             # Orange for options
+    "uri": "#00aaaa",                # Cyan for URIs
+    "prompt": "#00aa00 bold",        # Green prompt
+})