codekingpro/portable-devtools

1"""A simple SQLite CLI for the sqlite3 module.2 3Apart from using 'argparse' for the command-line interface,4this module implements the REPL as a thin wrapper around5the InteractiveConsole class from the 'code' stdlib module.6"""7import sqlite38import sys9 10from argparse import ArgumentParser11from code import InteractiveConsole12from textwrap import dedent13 14 15def execute(c, sql, suppress_errors=True):16    """Helper that wraps execution of SQL code.17 18    This is used both by the REPL and by direct execution from the CLI.19 20    'c' may be a cursor or a connection.21    'sql' is the SQL string to execute.22    """23 24    try:25        for row in c.execute(sql):26            print(row)27    except sqlite3.Error as e:28        tp = type(e).__name__29        try:30            print(f"{tp} ({e.sqlite_errorname}): {e}", file=sys.stderr)31        except AttributeError:32            print(f"{tp}: {e}", file=sys.stderr)33        if not suppress_errors:34            sys.exit(1)35 36 37class SqliteInteractiveConsole(InteractiveConsole):38    """A simple SQLite REPL."""39 40    def __init__(self, connection):41        super().__init__()42        self._con = connection43        self._cur = connection.cursor()44 45    def runsource(self, source, filename="<input>", symbol="single"):46        """Override runsource, the core of the InteractiveConsole REPL.47 48        Return True if more input is needed; buffering is done automatically.49        Return False if input is a complete statement ready for execution.50        """51        if not source or source.isspace():52            return False53        if source[0] == ".":54            match source[1:].strip():55                case "version":56                    print(f"{sqlite3.sqlite_version}")57                case "help":58                    print("Enter SQL code and press enter.")59                case "quit":60                    sys.exit(0)61                case "":62                    pass63                case _ as unknown:64                    self.write("Error: unknown command or invalid arguments:"65                               f'  "{unknown}".\n')66        else:67            if not sqlite3.complete_statement(source):68                return True69            execute(self._cur, source)70        return False71 72 73def main(*args):74    parser = ArgumentParser(75        description="Python sqlite3 CLI",76        color=True,77    )78    parser.add_argument(79        "filename", type=str, default=":memory:", nargs="?",80        help=(81            "SQLite database to open (defaults to ':memory:'). "82            "A new database is created if the file does not previously exist."83        ),84    )85    parser.add_argument(86        "sql", type=str, nargs="?",87        help=(88            "An SQL query to execute. "89            "Any returned rows are printed to stdout."90        ),91    )92    parser.add_argument(93        "-v", "--version", action="version",94        version=f"SQLite version {sqlite3.sqlite_version}",95        help="Print underlying SQLite library version",96    )97    args = parser.parse_args(*args)98 99    if args.filename == ":memory:":100        db_name = "a transient in-memory database"101    else:102        db_name = repr(args.filename)103 104    # Prepare REPL banner and prompts.105    if sys.platform == "win32" and "idlelib.run" not in sys.modules:106        eofkey = "CTRL-Z"107    else:108        eofkey = "CTRL-D"109    banner = dedent(f"""110        sqlite3 shell, running on SQLite version {sqlite3.sqlite_version}111        Connected to {db_name}112 113        Each command will be run using execute() on the cursor.114        Type ".help" for more information; type ".quit" or {eofkey} to quit.115    """).strip()116    sys.ps1 = "sqlite> "117    sys.ps2 = "    ... "118 119    con = sqlite3.connect(args.filename, isolation_level=None)120    try:121        if args.sql:122            # SQL statement provided on the command-line; execute it directly.123            execute(con, args.sql, suppress_errors=False)124        else:125            # No SQL provided; start the REPL.126            console = SqliteInteractiveConsole(con)127            try:128                import readline  # noqa: F401129            except ImportError:130                pass131            console.interact(banner, exitmsg="")132    finally:133        con.close()134 135    sys.exit(0)136 137 138if __name__ == "__main__":139    main(sys.argv[1:])140