Add JSON output for RedisVL CLI read commands (#593)

### Changes
This PR adds JSON output for RedisVL CLI read commands to make it more
machine-readable.

Features added:
- added the `--json` flag to `rvl version`, `rvl index listall`, `rvl
index info`, and `rvl stats`.
- Add shared JSON CLI helpers (`add_json_output_flag`, `cli_print_json`)
in `redisvl/cli/utils.py` for consistent one-object stdout behavior.
- Add/update unit tests for all new JSON paths and error contracts.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Adds optional output formatting for CLI read commands without changing
index/data operations. Main risk is minor: downstream scripts may depend
on exact stdout text when `--json` is used, but default behavior is
preserved.
> 
> **Overview**
> Adds a shared CLI JSON output mode via `--json`, including new helpers
`add_json_output_flag` and `cli_print_json` (with safe `bytes` encoding)
to ensure **one JSON object on stdout** on success.
> 
> Extends `rvl index info`, `rvl index listall`, and `rvl stats` to
support JSON output: `index info` now normalizes `FT.INFO`-style
structures into a stable `{index_information, index_fields}` payload,
`listall` emits `{indices: [...]}`, and `stats` emits a full
`STATS_KEYS`-shaped object with missing keys as `null`, while keeping
existing human-readable output when `--json` is not used.
> 
> Adds unit tests covering JSON vs table behavior, key
ordering/normalization, bytes decoding, and ensuring **no partial JSON
is printed on error paths**.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
e1d2bcf. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Vishal Bala <vishal-bala@users.noreply.github.com>

Author: limjoobin

redisvl/cli/index.py

@@ -5,7 +5,12 @@
 import yaml
 from pydantic import ValidationError
 
-from redisvl.cli.utils import add_index_parsing_options, create_redis_url
+from redisvl.cli.utils import (
+    add_index_parsing_options,
+    add_json_output_flag,
+    cli_print_json,
+    create_redis_url,
+)
 from redisvl.exceptions import RedisSearchError
 from redisvl.index import SearchIndex
 from redisvl.redis.connection import RedisConnectionFactory
@@ -51,11 +56,11 @@ class Index:
         [
             "rvl index <command> [<args>]\n",
             "Commands:",
-            "\tinfo        Obtain information about an index",
+            "\tinfo        Obtain information about an index (use --json for machine output)",
             "\tcreate      Create a new index",
             "\tdelete      Delete an existing index",
             "\tdestroy     Delete an existing index and all of its data",
-            "\tlistall     List all indexes",
+            "\tlistall     List all indexes (use --json for machine output)",
             "\n",
         ]
     )
@@ -65,6 +70,7 @@ def __init__(self):
 
         parser.add_argument("command", help="Subcommand to run")
         parser = add_index_parsing_options(parser)
+        parser = add_json_output_flag(parser)
 
         args = parser.parse_args(sys.argv[2:])
 
@@ -105,26 +111,34 @@ def info(self, args: Namespace):
         """Obtain information about an index.
 
         Usage:
-            rvl index info -i <index_name> | -s <schema_path>
+            rvl index info -i <index_name> | -s <schema_path> [--json]
         """
         index = self._connect_to_index(args)
         try:
-            _display_in_table(index.info())
+            index_info = index.info()
         except RedisSearchError as e:
             exit_redis_search_error(args, index, e)
 
+        if args.json:
+            cli_print_json(_index_info_for_json(index_info))
+        else:
+            _display_in_table(index_info)
+
     def listall(self, args: Namespace):
         """List all indices.
 
         Usage:
-            rvl index listall
+            rvl index listall [--json]
         """
         redis_url = create_redis_url(args)
         conn = RedisConnectionFactory.get_redis_connection(redis_url=redis_url)
         indices = convert_bytes(conn.execute_command("FT._LIST"))
-        print("Indices:")
-        for i, index in enumerate(indices):
-            print(str(i + 1) + ". " + index)
+        if args.json:
+            cli_print_json({"indices": indices})
+        else:
+            print("Indices:")
+            for i, index in enumerate(indices, start=1):
+                print(str(i) + ". " + index)
 
     def delete(self, args: Namespace, drop=False):
         """Delete an index.
@@ -167,6 +181,56 @@ def _connect_to_index(self, args: Namespace) -> SearchIndex:
         sys.exit(2)
 
 
+def _index_info_for_json(index_info: dict) -> dict:
+    """Build the JSON payload from the same fields shown in table mode."""
+    definition_src = index_info.get("index_definition")
+    if isinstance(definition_src, list):
+        definition = convert_bytes(make_dict(definition_src))
+    elif isinstance(definition_src, tuple):
+        definition = convert_bytes(make_dict(list(definition_src)))
+    elif isinstance(definition_src, dict):
+        definition = convert_bytes(dict(definition_src))
+    else:
+        definition = {}
+    attributes = index_info.get("attributes", [])
+    index_fields = []
+
+    for attrs in attributes:
+        if isinstance(attrs, list):
+            attr = convert_bytes(make_dict(attrs))
+        elif isinstance(attrs, tuple):
+            attr = convert_bytes(make_dict(list(attrs)))
+        elif isinstance(attrs, dict):
+            attr = convert_bytes(dict(attrs))
+        else:
+            attr = {}
+        field = {
+            "name": attr.get("identifier"),
+            "attribute": attr.get("attribute"),
+            "type": attr.get("type"),
+        }
+        field_options = {
+            k: v
+            for k, v in attr.items()
+            if k not in {"identifier", "attribute", "type"}
+        }
+        if field_options:
+            field["field_options"] = field_options
+        index_fields.append(field)
+
+    payload = {
+        "index_information": {
+            "index_name": index_info.get("index_name"),
+            "storage_type": definition.get("key_type"),
+            "prefixes": definition.get("prefixes"),
+            "index_options": index_info.get("index_options"),
+            "indexing": index_info.get("indexing"),
+        },
+        "index_fields": index_fields,
+    }
+    return convert_bytes(payload)
+
+
 def _display_in_table(index_info):
     print("\n")
     attributes = index_info.get("attributes", [])

redisvl/cli/stats.py

@@ -5,7 +5,12 @@
 import yaml
 from pydantic import ValidationError
 
-from redisvl.cli.utils import add_index_parsing_options, create_redis_url
+from redisvl.cli.utils import (
+    add_index_parsing_options,
+    add_json_output_flag,
+    cli_print_json,
+    create_redis_url,
+)
 from redisvl.exceptions import RedisSearchError
 from redisvl.index import SearchIndex
 from redisvl.schema.schema import IndexSchema
@@ -67,6 +72,17 @@ def exit_redis_search_error(
 ]
 
 
+def _stats_rows(index_info: dict) -> list[tuple[str, object]]:
+    """Normalize ``index.info()`` for both JSON and table output.
+
+    Returns ordered ``(key, value)`` pairs: keys follow ``STATS_KEYS``, values
+    preserve native types from ``index_info``. Missing keys remain ``None``.
+    For JSON, wrap the result in ``dict(...)``; the table stringifies values
+    when rendering.
+    """
+    return [(key, index_info.get(key)) for key in STATS_KEYS]
+
+
 class Stats:
     usage = "\n".join(
         [
@@ -77,6 +93,7 @@ class Stats:
     def __init__(self):
         parser = argparse.ArgumentParser(usage=self.usage)
         parser = add_index_parsing_options(parser)
+        parser = add_json_output_flag(parser)
         args = parser.parse_args(sys.argv[2:])
 
         try:
@@ -94,10 +111,16 @@ def stats(self, args: Namespace):
         """
         index = self._connect_to_index(args)
         try:
-            _display_stats(index.info())
+            index_info = index.info()
         except RedisSearchError as e:
             exit_redis_search_error(args, index, e)
 
+        rows = _stats_rows(index_info)
+        if args.json:
+            cli_print_json(dict(rows))
+        else:
+            _display_stats(rows)
+
     def _connect_to_index(self, args: Namespace) -> SearchIndex:
         redis_url = create_redis_url(args)
 
@@ -118,10 +141,7 @@ def _connect_to_index(self, args: Namespace) -> SearchIndex:
         sys.exit(2)
 
 
-def _display_stats(index_info):
-    # Extracting the statistics
-    stats_data = [(key, str(index_info.get(key))) for key in STATS_KEYS]
-
+def _display_stats(stats_data: list[tuple[str, object]]) -> None:
     # Display the statistics in tabular format
     print("\nStatistics:")
     max_key_length = max(len(key) for key, _ in stats_data)
@@ -130,5 +150,6 @@ def _display_stats(index_info):
     print("│ Stat Key                    │ Value      │")  # header row
     print(f"├{horizontal_line}┼────────────┤")  # separator row
     for key, value in stats_data:
-        print(f"│ {key:<27} │ {value[0:10]:<10} │")  # data rows
+        value_str = str(value)
+        print(f"│ {key:<27} │ {value_str[0:10]:<10} │")  # data rows
     print(f"╰{horizontal_line}┴────────────╯")  # bottom row

redisvl/cli/utils.py

@@ -1,5 +1,7 @@
+import json
 import os
 from argparse import ArgumentParser, Namespace
+from typing import Any, Mapping
 from urllib.parse import quote, urlparse, urlunparse
 
 from redisvl.redis.constants import REDIS_URL_ENV_VAR
@@ -91,3 +93,27 @@ def add_index_parsing_options(parser: ArgumentParser) -> ArgumentParser:
         default=None,
     )
     return parser
+
+
+def _cli_json_default(obj: object) -> object:
+    """Handle common non-JSON-native types (e.g. bytes from Redis) for cli_print_json."""
+    if isinstance(obj, bytes):
+        return obj.decode("utf-8", errors="replace")
+    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
+
+
+def add_json_output_flag(parser: ArgumentParser) -> ArgumentParser:
+    """Register ``--json`` for machine-readable stdout (one JSON object on success)."""
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        dest="json",
+        default=False,
+        help="Format output (when successful) as machine-readable JSON",
+    )
+    return parser
+
+
+def cli_print_json(data: Mapping[str, Any]) -> None:
+    """Write a single JSON object to stdout (deterministic key order for tests and scripts)."""
+    print(json.dumps(data, default=_cli_json_default))