🪲TypeError: expected string or bytes-like object, got 'ArgumentInfo' #…

…2091

opteryx/__main__.py

@@ -16,21 +16,21 @@
 A command line interface for Opteryx
 """
 
+import argparse
 import os
 import readline
 import sys
 import threading
 import time
 
-import typer
-
 import opteryx
 from opteryx.exceptions import MissingSqlStatement
 from opteryx.utils.sql import clean_statement
 from opteryx.utils.sql import remove_comments
 
 sys.path.insert(1, os.path.join(sys.path[0], ".."))
 
+
 if readline:
     pass
 
@@ -55,148 +55,161 @@ def print_dots(stop_event):
         if not stop_event.is_set():
             print("\r   \r", end="", flush=True)
             time.sleep(0.5)
-        if not stop_event.is_set():
-            print("\r   \r", end="", flush=True)
-            time.sleep(0.5)
-        if not stop_event.is_set():
-            print("\r   \r", end="", flush=True)
-            time.sleep(0.5)
 
 
-# fmt:off
-def main(
-    o: str = typer.Option(default="console", help="Output location (ignored by REPL)", ),
-    color: bool = typer.Option(default=True, help="Colorize the table displayed to the console."),
-    table_width: bool = typer.Option(default=True, help="Limit console display to the screen width."),
-    max_col_width: int = typer.Option(default=64, help="Maximum column width"),
-    stats: bool = typer.Option(default=True, help="Report statistics."),
-    cycles: int = typer.Option(default=1, help="Repeat Execution."),
-    sql: str = typer.Argument(None, show_default=False, help="Execute SQL statement and quit."),
-):
-# fmt:on
-    """
-    Opteryx CLI
-    """
-    if hasattr(max_col_width, "default"):
-        max_col_width = max_col_width.default
-    if hasattr(table_width, "default"):
-        table_width = table_width.default
-    if hasattr(cycles, "default"):
-        cycles = cycles.default
-
-    if sql is None:  # pragma: no cover
-
-
-        if o != "console":
+def main():
+    parser = argparse.ArgumentParser(description="A command line interface for Opteryx")
+
+    parser.add_argument(
+        "--o", type=str, default="console", help="Output location (ignored by REPL)", dest="output"
+    )
+
+    # Mutually exclusive group for `--color` and `--no-color`
+    color_group = parser.add_mutually_exclusive_group()
+    color_group.add_argument(
+        "--color", dest="color", action="store_true", default=True, help="Colorize the table."
+    )
+    color_group.add_argument(
+        "--no-color", dest="color", action="store_false", help="Disable colorized output."
+    )
+
+    parser.add_argument(
+        "--table_width",
+        action="store_true",
+        default=True,
+        help="Limit console display to the screen width.",
+    )
+    parser.add_argument("--max_col_width", type=int, default=64, help="Maximum column width")
+
+    # Mutually exclusive group for `--color` and `--no-color`
+    stats_group = parser.add_mutually_exclusive_group()
+    stats_group.add_argument(
+        "--stats", dest="stats", action="store_true", default=True, help="Report statistics."
+    )
+    stats_group.add_argument(
+        "--no-stats", dest="stats", action="store_false", help="Disable report statistics."
+    )
+
+    parser.add_argument("--cycles", type=int, default=1, help="Repeat Execution.")
+    parser.add_argument("sql", type=str, nargs="?", help="Execute SQL statement and quit.")
+
+    args = parser.parse_args()
+
+    # Run in REPL mode if no SQL is provided
+    if args.sql is None:  # pragma: no cover
+        if args.output != "console":
             raise ValueError("Cannot specify output location and not provide a SQL statement.")
-
         print(f"Opteryx version {opteryx.__version__}")
         print("  Enter '.help' for usage hints")
         print("  Enter '.exit' to exit this program")
 
-        # Start the REPL loop
-        while True:  # pragma: no cover
-            # Prompt the user for a SQL statement
+        while True:  # REPL loop
             print()
-            statement = input('opteryx> ')
-
-            # If the user entered "exit", exit the loop
-            # forgive them for 'quit'
-            if statement in {'.exit', '.quit'}:
+            statement = input("opteryx> ")
+            if statement in {".exit", ".quit"}:
                 break
             if statement == ".help":
                 print("  .exit        Exit this program")
                 print("  .help        Show help text")
                 continue
 
-            # Create a stop event
             stop_event = threading.Event()
-            # Create and start a thread to print dots
             dot_thread = threading.Thread(target=print_dots, args=(stop_event,))
             dot_thread.start()
             try:
-                # Execute the SQL statement and display the results
                 start = time.monotonic_ns()
                 result = opteryx.query(statement, memberships=["opteryx"])
                 result.materialize()
                 stop_event.set()
                 duration = time.monotonic_ns() - start
                 print("\r   \r", end="", flush=True)
-                print(result.display(limit=-1, display_width=table_width, colorize=color, max_column_width=max_col_width))
-                if stats:
-                    print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
+                print(
+                    result.display(
+                        limit=-1,
+                        display_width=args.table_width,
+                        colorize=args.color,
+                        max_column_width=args.max_col_width,
+                    )
+                )
+                if args.stats:
+                    print(
+                        f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )"
+                    )
             except MissingSqlStatement:
-                print("\r   \r", end="", flush=True)
-                print(f"{ANSI_RED}Error{ANSI_RESET}: Expected SQL statement or dot command missing.")
-                print("  Enter '.help' for usage hints")
+                print(
+                    f"{ANSI_RED}Error{ANSI_RESET}: Expected SQL statement or dot command missing."
+                )
             except Exception as e:
-                print("\r   \r", end="", flush=True)
-                # Display a friendly error message if an exception occurs
                 print(f"{ANSI_RED}Error{ANSI_RESET}: {e}")
-                print("  Enter '.help' for usage hints")
             finally:
-                # Stop the dot thread
                 stop_event.set()
                 dot_thread.join()
         quit()
 
-    # tidy up the statement
-    sql = clean_statement(remove_comments(sql))
+    # Process the SQL query
+    sql = clean_statement(remove_comments(args.sql))
 
-    if cycles > 1:  # pragma: no cover
-        # this is useful for benchmarking
+    if args.cycles > 1:  # Benchmarking mode
         print("[", end="")
-        for i in range(cycles):
+        for i in range(args.cycles):
             start = time.monotonic_ns()
             result = opteryx.query_to_arrow(sql)
-            print((time.monotonic_ns() - start) / 1e9, flush=True, end=("," if (i+1) < cycles else "]\n"))
+            print(
+                (time.monotonic_ns() - start) / 1e9,
+                flush=True,
+                end=("," if (i + 1) < args.cycles else "]\n"),
+            )
         return
 
     start = time.monotonic_ns()
     result = opteryx.query(sql)
     result.materialize()
     duration = time.monotonic_ns() - start
 
-    if o == "console":
-        print(result.display(limit=-1, display_width=table_width, colorize=color, max_column_width=max_col_width))
-        if stats:
-            print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
+    if args.output == "console":
+        print(
+            result.display(
+                limit=-1,
+                display_width=args.table_width,
+                colorize=args.color,
+                max_column_width=args.max_col_width,
+            )
+        )
+        if args.stats:
+            print(
+                f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )"
+            )
     else:
         table = result.arrow()
-
-        ext = o.lower().split(".")[-1]
+        ext = args.output.lower().split(".")[-1]
 
         if ext == "parquet":
             from pyarrow import parquet
 
-            parquet.write_table(table, o)
-            print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
-            print(f"Written result to '{o}'")
+            parquet.write_table(table, args.output)
         elif ext == "csv":
             from pyarrow import csv
 
-            csv.write_csv(table, o)
-            print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
-            print(f"Written result to '{o}'")
+            csv.write_csv(table, args.output)
         elif ext == "jsonl":
             import orjson
-            with open(o, mode="wb") as file:
+
+            with open(args.output, mode="wb") as file:
                 for row in result:
                     file.write(orjson.dumps(row.as_dict, default=str) + b"\n")
-            print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
-            print(f"Written result to '{o}'")
         elif ext == "md":
-            with open(o, mode="w") as file:
+            with open(args.output, mode="w") as file:
                 file.write(result.markdown(limit=-1))
-            print(f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )")
-            print(f"Written result to '{o}'")
         else:
-            raise ValueError(f"Unknown output format '{ext}'")  # pragma: no cover
+            raise ValueError(f"Unknown output format '{ext}'")
+        print(
+            f"[ {result.rowcount} rows x {result.columncount} columns ] ( {duration/1e9} seconds )"
+        )
+        print(f"Written result to '{args.output}'")
 
 
-if __name__ == "__main__":  # pragma: no cover
+if __name__ == "__main__":
     try:
-        typer.run(main)
+        main()
     except Exception as e:
-        # Display a friendly error message if an exception occurs
         print(f"{ANSI_RED}Error{ANSI_RESET}: {e}")

opteryx/__version__.py

@@ -28,7 +28,7 @@ class VersionStatus(Enum):
 
 _major = 0
 _minor = 18
-_revision = 1
+_revision = 2
 _status = VersionStatus.RELEASE
 
 __author__ = "@joocer"

opteryx/command.py

@@ -14,4 +14,4 @@
 
 from opteryx.__main__ import main
 
-__all__ = "main"
+__all__ = ("main",)

tests/misc/test_cli.py

@@ -1,26 +1,73 @@
-"""
-Touch Test Only:
-We test the CLI runs when we touch it, we're not checking the outputs.
-"""
-
 import os
 import sys
+import subprocess
 
 sys.path.insert(1, os.path.join(sys.path[0], "../.."))
 
-from opteryx.command import main
+def run_cli(args):
+    """Helper function to run the CLI and return the result."""
+    result = subprocess.run(
+        [sys.executable, "opteryx"] + args,
+        capture_output=True,
+        text=True,
+        timeout=5
+    )
+    return result
+
+
+def test_basic_execution():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["SELECT * FROM $planets"])
+    assert result.returncode == 0
+
+def test_save_to_file():
+    result = run_cli(["--o", "planets.parquet", "SELECT * FROM $planets"])
+
+    assert result.returncode == 0, result.stderr
+    assert "Written result to" in result.stdout
+
+def test_colorized():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--color", "SELECT * FROM $planets"])
+    assert result.returncode == 0
+
+def test_decolorized():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--no-color", "SELECT * FROM $planets"])
+    assert result.returncode == 0
+
+def test_stats():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--stats", "SELECT * FROM $planets"])
+    assert result.returncode == 0
+
+def test_no_stats():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--no-stats", "SELECT * FROM $planets"])
+    assert result.returncode == 0
+
+def test_cycles():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--cycles", "3", "SELECT * FROM $planets"])
+    assert result.returncode == 0
 
+def test_table_width():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--table_width", "SELECT * FROM $planets"])
+    assert result.returncode == 0
 
-def test_basic_cli():
-    main(sql="SELECT * FROM $planets;", o="console", max_col_width=5)
-    main(sql="SELECT * FROM $planets;", o="console")
-    main(sql="SELECT * FROM $planets;", o="temp.csv")
-    main(sql="SELECT * FROM $planets;", o="temp.jsonl")
-    main(sql="SELECT * FROM $planets;", o="temp.parquet")
-    main(sql="SELECT * FROM $planets;", o="temp.md")
+def test_column_width():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--no-color", "--max_col_width", "4", "SELECT * FROM $planets"])
+    assert result.returncode == 0
+    assert '│ Merc │' in result.stdout
 
+def test_unknown_param():
+    """Test the CLI when no SQL is provided, expecting an error."""
+    result = run_cli(["--verbose", "SELECT * FROM $planets"])
+    assert result.returncode != 0
 
 if __name__ == "__main__":  # pragma: no cover
-    test_basic_cli()
+    from tests.tools import run_tests
 
-    print("✅ okay")
+    run_tests()