Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>
- Loading branch information
1 parent
1e6b635
commit bc7c7cd
Showing
5 changed files
with
281 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,97 @@ | ||
import sqlite3 | ||
import sys | ||
|
||
from argparse import ArgumentParser | ||
from code import InteractiveConsole | ||
from textwrap import dedent | ||
|
||
|
||
def execute(c, sql, suppress_errors=True): | ||
try: | ||
for row in c.execute(sql): | ||
print(row) | ||
except sqlite3.Error as e: | ||
tp = type(e).__name__ | ||
try: | ||
print(f"{tp} ({e.sqlite_errorname}): {e}", file=sys.stderr) | ||
except AttributeError: | ||
print(f"{tp}: {e}", file=sys.stderr) | ||
if not suppress_errors: | ||
sys.exit(1) | ||
|
||
|
||
class SqliteInteractiveConsole(InteractiveConsole): | ||
|
||
def __init__(self, connection): | ||
super().__init__() | ||
self._con = connection | ||
self._cur = connection.cursor() | ||
|
||
def runsource(self, source, filename="<input>", symbol="single"): | ||
match source: | ||
case ".version": | ||
print(f"{sqlite3.sqlite_version}") | ||
case ".help": | ||
print("Enter SQL code and press enter.") | ||
case ".quit": | ||
sys.exit(0) | ||
case _: | ||
if not sqlite3.complete_statement(source): | ||
return True | ||
execute(self._cur, source) | ||
return False | ||
|
||
|
||
def main(): | ||
parser = ArgumentParser( | ||
description="Python sqlite3 CLI", | ||
prog="python -m sqlite3", | ||
) | ||
parser.add_argument( | ||
"filename", type=str, default=":memory:", nargs="?", | ||
help=( | ||
"SQLite database to open (defaults to ':memory:'). " | ||
"A new database is created if the file does not previously exist." | ||
), | ||
) | ||
parser.add_argument( | ||
"sql", type=str, nargs="?", | ||
help=( | ||
"An SQL query to execute. " | ||
"Any returned rows are printed to stdout." | ||
), | ||
) | ||
parser.add_argument( | ||
"-v", "--version", action="version", | ||
version=f"SQLite version {sqlite3.sqlite_version}", | ||
help="Print underlying SQLite library version", | ||
) | ||
args = parser.parse_args() | ||
|
||
if args.filename == ":memory:": | ||
db_name = "a transient in-memory database" | ||
else: | ||
db_name = repr(args.filename) | ||
|
||
banner = dedent(f""" | ||
sqlite3 shell, running on SQLite version {sqlite3.sqlite_version} | ||
Connected to {db_name} | ||
Each command will be run using execute() on the cursor. | ||
Type ".help" for more information; type ".quit" or CTRL-D to quit. | ||
""").strip() | ||
sys.ps1 = "sqlite> " | ||
sys.ps2 = " ... " | ||
|
||
con = sqlite3.connect(args.filename, isolation_level=None) | ||
try: | ||
if args.sql: | ||
execute(con, args.sql, suppress_errors=False) | ||
else: | ||
console = SqliteInteractiveConsole(con) | ||
console.interact(banner, exitmsg="") | ||
finally: | ||
con.close() | ||
|
||
|
||
main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,155 @@ | ||
"""sqlite3 CLI tests.""" | ||
|
||
import sqlite3 as sqlite | ||
import subprocess | ||
import sys | ||
import unittest | ||
|
||
from test.support import SHORT_TIMEOUT, requires_subprocess | ||
from test.support.os_helper import TESTFN, unlink | ||
|
||
|
||
@requires_subprocess() | ||
class CommandLineInterface(unittest.TestCase): | ||
|
||
def _do_test(self, *args, expect_success=True): | ||
with subprocess.Popen( | ||
[sys.executable, "-Xutf8", "-m", "sqlite3", *args], | ||
encoding="utf-8", | ||
bufsize=0, | ||
stdout=subprocess.PIPE, | ||
stderr=subprocess.PIPE, | ||
) as proc: | ||
proc.wait() | ||
if expect_success == bool(proc.returncode): | ||
self.fail("".join(proc.stderr)) | ||
stdout = proc.stdout.read() | ||
stderr = proc.stderr.read() | ||
if expect_success: | ||
self.assertEqual(stderr, "") | ||
else: | ||
self.assertEqual(stdout, "") | ||
return stdout, stderr | ||
|
||
def expect_success(self, *args): | ||
out, _ = self._do_test(*args) | ||
return out | ||
|
||
def expect_failure(self, *args): | ||
_, err = self._do_test(*args, expect_success=False) | ||
return err | ||
|
||
def test_cli_help(self): | ||
out = self.expect_success("-h") | ||
self.assertIn("usage: python -m sqlite3", out) | ||
|
||
def test_cli_version(self): | ||
out = self.expect_success("-v") | ||
self.assertIn(sqlite.sqlite_version, out) | ||
|
||
def test_cli_execute_sql(self): | ||
out = self.expect_success(":memory:", "select 1") | ||
self.assertIn("(1,)", out) | ||
|
||
def test_cli_execute_too_much_sql(self): | ||
stderr = self.expect_failure(":memory:", "select 1; select 2") | ||
err = "ProgrammingError: You can only execute one statement at a time" | ||
self.assertIn(err, stderr) | ||
|
||
def test_cli_execute_incomplete_sql(self): | ||
stderr = self.expect_failure(":memory:", "sel") | ||
self.assertIn("OperationalError (SQLITE_ERROR)", stderr) | ||
|
||
def test_cli_on_disk_db(self): | ||
self.addCleanup(unlink, TESTFN) | ||
out = self.expect_success(TESTFN, "create table t(t)") | ||
self.assertEqual(out, "") | ||
out = self.expect_success(TESTFN, "select count(t) from t") | ||
self.assertIn("(0,)", out) | ||
|
||
|
||
@requires_subprocess() | ||
class InteractiveSession(unittest.TestCase): | ||
TIMEOUT = SHORT_TIMEOUT / 10. | ||
MEMORY_DB_MSG = "Connected to a transient in-memory database" | ||
PS1 = "sqlite> " | ||
PS2 = "... " | ||
|
||
def start_cli(self, *args): | ||
return subprocess.Popen( | ||
[sys.executable, "-Xutf8", "-m", "sqlite3", *args], | ||
encoding="utf-8", | ||
bufsize=0, | ||
stdin=subprocess.PIPE, | ||
# Note: the banner is printed to stderr, the prompt to stdout. | ||
stdout=subprocess.PIPE, | ||
stderr=subprocess.PIPE, | ||
) | ||
|
||
def expect_success(self, proc): | ||
proc.wait() | ||
if proc.returncode: | ||
self.fail("".join(proc.stderr)) | ||
|
||
def test_interact(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn(self.PS1, out) | ||
self.expect_success(proc) | ||
|
||
def test_interact_quit(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(input=".quit", timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn(self.PS1, out) | ||
self.expect_success(proc) | ||
|
||
def test_interact_version(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(input=".version", timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn(sqlite.sqlite_version, out) | ||
self.expect_success(proc) | ||
|
||
def test_interact_valid_sql(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(input="select 1;", | ||
timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn("(1,)", out) | ||
self.expect_success(proc) | ||
|
||
def test_interact_valid_multiline_sql(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(input="select 1\n;", | ||
timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn(self.PS2, out) | ||
self.assertIn("(1,)", out) | ||
self.expect_success(proc) | ||
|
||
def test_interact_invalid_sql(self): | ||
with self.start_cli() as proc: | ||
out, err = proc.communicate(input="sel;", timeout=self.TIMEOUT) | ||
self.assertIn(self.MEMORY_DB_MSG, err) | ||
self.assertIn("OperationalError (SQLITE_ERROR)", err) | ||
self.expect_success(proc) | ||
|
||
def test_interact_on_disk_file(self): | ||
self.addCleanup(unlink, TESTFN) | ||
with self.start_cli(TESTFN) as proc: | ||
out, err = proc.communicate(input="create table t(t);", | ||
timeout=self.TIMEOUT) | ||
self.assertIn(TESTFN, err) | ||
self.assertIn(self.PS1, out) | ||
self.expect_success(proc) | ||
with self.start_cli(TESTFN, "select count(t) from t") as proc: | ||
out = proc.stdout.read() | ||
err = proc.stderr.read() | ||
self.assertIn("(0,)", out) | ||
self.expect_success(proc) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
2 changes: 2 additions & 0 deletions
2
Misc/NEWS.d/next/Library/2022-07-20-00-23-58.gh-issue-77617.XGaqSQ.rst
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,2 @@ | ||
Add :mod:`sqlite3` :ref:`command-line interface <sqlite3-cli>`. | ||
Patch by Erlend Aasland. |