Fix/cli: standardize exit codes, messages and streams (#592)

hillarytoh · web-flow · commit fc9ccbd15f1f · 2026-04-30T10:11:42.000+08:00
- Standardize non-MCP rvl CLI behavior: exit 0 for success/help, exit 2
for usage/argument and schema input issues, exit 1 for runtime failures;
stderr for errors and stdout for normal output or help.

- Catch specific exception types in the CLI to map to the right exit
code (e.g. invalid path vs Redis search)

- Version: print version to stdout (not only via logger).
diff --git a/redisvl/cli/index.py b/redisvl/cli/index.py
@@ -2,7 +2,11 @@
 import sys
 from argparse import Namespace
 
+import yaml
+from pydantic import ValidationError
+
 from redisvl.cli.utils import add_index_parsing_options, create_redis_url
+from redisvl.exceptions import RedisSearchError
 from redisvl.index import SearchIndex
 from redisvl.redis.connection import RedisConnectionFactory
 from redisvl.redis.utils import convert_bytes, make_dict
@@ -11,6 +15,36 @@
 
 logger = get_logger("[RedisVL]")
 
+# Exceptions commonly raised when loading or validating a schema path (-s).
+SCHEMA_INPUT_ERRORS = (
+    FileNotFoundError,
+    ValueError,
+    yaml.YAMLError,
+    ValidationError,
+)
+
+
+def exit_schema_input_error(args: Namespace, exc: BaseException) -> None:
+    if not args.schema:
+        raise exc
+    print(str(exc), file=sys.stderr)
+    sys.exit(2)
+
+
+def exit_redis_search_error(
+    args: Namespace, index: SearchIndex | None, exc: RedisSearchError
+) -> None:
+    name = (
+        index.schema.index.name
+        if index is not None
+        else (args.index or args.schema or "unknown")
+    )
+    print(
+        f"Redis search operation failed for index {name!r}. {exc}",
+        file=sys.stderr,
+    )
+    sys.exit(1)
+
 
 class Index:
     usage = "\n".join(
@@ -33,26 +67,38 @@ def __init__(self):
         parser = add_index_parsing_options(parser)
 
         args = parser.parse_args(sys.argv[2:])
+
         if not hasattr(self, args.command):
-            parser.print_help()
-            exit(0)
+            print(f"Unknown command: {args.command}\n", file=sys.stderr)
+            parser.print_help(sys.stderr)
+            sys.exit(2)
+
         try:
             getattr(self, args.command)(args)
         except Exception as e:
-            logger.error(e)
-            exit(0)
+            logger.error(e, exc_info=True)
+            print(str(e), file=sys.stderr)
+            sys.exit(1)
 
     def create(self, args: Namespace):
         """Create an index.
 
         Usage:
-            rvl index create -i <index_name> | -s <schema_path>
+            rvl index create -s <schema_path>
         """
         if not args.schema:
-            print("Schema must be provided to create an index")
-            exit(1)
-        index = SearchIndex.from_yaml(args.schema, redis_url=create_redis_url(args))
-        index.create()
+            print("Schema must be provided to create an index", file=sys.stderr)
+            sys.exit(2)
+
+        redis_url = create_redis_url(args)
+        try:
+            index = SearchIndex.from_yaml(args.schema, redis_url=redis_url)
+        except SCHEMA_INPUT_ERRORS as e:
+            exit_schema_input_error(args, e)
+        try:
+            index.create()
+        except RedisSearchError as e:
+            exit_redis_search_error(args, index, e)
         print("Index created successfully")
 
     def info(self, args: Namespace):
@@ -62,7 +108,10 @@ def info(self, args: Namespace):
             rvl index info -i <index_name> | -s <schema_path>
         """
         index = self._connect_to_index(args)
-        _display_in_table(index.info())
+        try:
+            _display_in_table(index.info())
+        except RedisSearchError as e:
+            exit_redis_search_error(args, index, e)
 
     def listall(self, args: Namespace):
         """List all indices.
@@ -84,7 +133,10 @@ def delete(self, args: Namespace, drop=False):
             rvl index delete -i <index_name> | -s <schema_path>
         """
         index = self._connect_to_index(args)
-        index.delete(drop=drop)
+        try:
+            index.delete(drop=drop)
+        except RedisSearchError as e:
+            exit_redis_search_error(args, index, e)
         print("Index deleted successfully")
 
     def destroy(self, args: Namespace):
@@ -96,19 +148,23 @@ def destroy(self, args: Namespace):
         self.delete(args, drop=True)
 
     def _connect_to_index(self, args: Namespace) -> SearchIndex:
-        # connect to redis
         redis_url = create_redis_url(args)
 
         if args.index:
-            schema = IndexSchema.from_dict({"index": {"name": args.index}})
-            index = SearchIndex(schema=schema, redis_url=redis_url)
-        elif args.schema:
-            index = SearchIndex.from_yaml(args.schema, redis_url=redis_url)
-        else:
-            print("Index name or schema must be provided")
-            exit(1)
-
-        return index
+            try:
+                schema = IndexSchema.from_dict({"index": {"name": args.index}})
+                return SearchIndex(schema=schema, redis_url=redis_url)
+            except SCHEMA_INPUT_ERRORS as e:
+                exit_schema_input_error(args, e)
+
+        if args.schema:
+            try:
+                return SearchIndex.from_yaml(args.schema, redis_url=redis_url)
+            except SCHEMA_INPUT_ERRORS as e:
+                exit_schema_input_error(args, e)
+
+        print("Index name or schema must be provided", file=sys.stderr)
+        sys.exit(2)
 
 
 def _display_in_table(index_info):
diff --git a/redisvl/cli/main.py b/redisvl/cli/main.py
@@ -30,29 +30,36 @@ def __init__(self):
         parser.add_argument("command", help="Subcommand to run")
 
         if len(sys.argv) < 2:
-            parser.print_help()
-            exit(0)
+            parser.print_help(sys.stdout)
+            sys.exit(0)
 
         args = parser.parse_args(sys.argv[1:2])
+
         if not hasattr(self, args.command):
-            parser.print_help()
-            exit(0)
-        getattr(self, args.command)()
+            print(f"Unknown command: {args.command}\n", file=sys.stderr)
+            parser.print_help(sys.stderr)
+            sys.exit(2)
+
+        try:
+            getattr(self, args.command)()
+        except Exception as e:
+            print(e, file=sys.stderr)
+            sys.exit(1)
 
     def index(self):
         Index()
-        exit(0)
+        sys.exit(0)
 
     def mcp(self):
         from redisvl.cli.mcp import MCP
 
         MCP()
-        exit(0)
+        sys.exit(0)
 
     def version(self):
         Version()
-        exit(0)
+        sys.exit(0)
 
     def stats(self):
         Stats()
-        exit(0)
+        sys.exit(0)
diff --git a/redisvl/cli/mcp.py b/redisvl/cli/mcp.py
@@ -7,11 +7,11 @@
 
 
 class _MCPArgumentParser(argparse.ArgumentParser):
-    """ArgumentParser variant that reports usage errors with exit code 1."""
+    """ArgumentParser variant that reports usage errors with exit code 2."""
 
     def error(self, message):
         self.print_usage(sys.stderr)
-        self.exit(1, "%s: error: %s\n" % (self.prog, message))
+        self.exit(2, "%s: error: %s\n" % (self.prog, message))
 
 
 class MCP:
diff --git a/redisvl/cli/stats.py b/redisvl/cli/stats.py
@@ -2,13 +2,48 @@
 import sys
 from argparse import Namespace
 
+import yaml
+from pydantic import ValidationError
+
 from redisvl.cli.utils import add_index_parsing_options, create_redis_url
+from redisvl.exceptions import RedisSearchError
 from redisvl.index import SearchIndex
 from redisvl.schema.schema import IndexSchema
 from redisvl.utils.log import get_logger
 
 logger = get_logger("[RedisVL]")
 
+# Exceptions commonly raised when loading or validating a schema path (-s).
+SCHEMA_INPUT_ERRORS = (
+    FileNotFoundError,
+    ValueError,
+    yaml.YAMLError,
+    ValidationError,
+)
+
+
+def exit_schema_input_error(args: Namespace, exc: BaseException) -> None:
+    if not args.schema:
+        raise exc
+    print(str(exc), file=sys.stderr)
+    sys.exit(2)
+
+
+def exit_redis_search_error(
+    args: Namespace, index: SearchIndex | None, exc: RedisSearchError
+) -> None:
+    name = (
+        index.schema.index.name
+        if index is not None
+        else (args.index or args.schema or "unknown")
+    )
+    print(
+        f"Redis search operation failed for index {name!r}. {exc}",
+        file=sys.stderr,
+    )
+    sys.exit(1)
+
+
 STATS_KEYS = [
     "num_docs",
     "num_terms",
@@ -43,11 +78,13 @@ def __init__(self):
         parser = argparse.ArgumentParser(usage=self.usage)
         parser = add_index_parsing_options(parser)
         args = parser.parse_args(sys.argv[2:])
+
         try:
             self.stats(args)
         except Exception as e:
-            logger.error(e)
-            exit(0)
+            logger.error(e, exc_info=True)
+            print(str(e), file=sys.stderr)
+            sys.exit(1)
 
     def stats(self, args: Namespace):
         """Obtain stats about an index.
@@ -56,22 +93,29 @@ def stats(self, args: Namespace):
             rvl stats -i <index_name> | -s <schema_path>
         """
         index = self._connect_to_index(args)
-        _display_stats(index.info())
+        try:
+            _display_stats(index.info())
+        except RedisSearchError as e:
+            exit_redis_search_error(args, index, e)
 
     def _connect_to_index(self, args: Namespace) -> SearchIndex:
-        # connect to redis
         redis_url = create_redis_url(args)
 
         if args.index:
-            schema = IndexSchema.from_dict({"index": {"name": args.index}})
-            index = SearchIndex(schema=schema, redis_url=redis_url)
-        elif args.schema:
-            index = SearchIndex.from_yaml(args.schema, redis_url=redis_url)
-        else:
-            logger.error("Index name or schema must be provided")
-            exit(0)
-
-        return index
+            try:
+                schema = IndexSchema.from_dict({"index": {"name": args.index}})
+                return SearchIndex(schema=schema, redis_url=redis_url)
+            except SCHEMA_INPUT_ERRORS as e:
+                exit_schema_input_error(args, e)
+
+        if args.schema:
+            try:
+                return SearchIndex.from_yaml(args.schema, redis_url=redis_url)
+            except SCHEMA_INPUT_ERRORS as e:
+                exit_schema_input_error(args, e)
+
+        print("Index name or schema must be provided", file=sys.stderr)
+        sys.exit(2)
 
 
 def _display_stats(index_info):
diff --git a/redisvl/cli/version.py b/redisvl/cli/version.py
@@ -3,9 +3,6 @@
 from argparse import Namespace
 
 from redisvl import __version__
-from redisvl.utils.log import get_logger
-
-logger = get_logger("[RedisVL]")
 
 
 class Version:
@@ -29,4 +26,4 @@ def version(self, args: Namespace):
         if args.short:
             print(__version__)
         else:
-            logger.info(f"RedisVL version {__version__}")
+            print(f"RedisVL version {__version__}")
diff --git a/tests/unit/test_cli_mcp.py b/tests/unit/test_cli_mcp.py
@@ -464,7 +464,7 @@ def test_mcp_command_rejects_invalid_transport(monkeypatch, capsys):
     with pytest.raises(SystemExit) as exc_info:
         module.MCP()
 
-    assert exc_info.value.code == 1
+    assert exc_info.value.code == 2
     out = capsys.readouterr()
     assert "invalid choice" in out.err or "invalid choice" in out.out
 
