#!/usr/bin/env python import sqlite3 import sys from pygments.lexers.sql import SqlLexer from prompt_toolkit import PromptSession from prompt_toolkit.completion import WordCompleter from prompt_toolkit.lexers import PygmentsLexer from prompt_toolkit.styles import Style sql_completer = WordCompleter( [ "abort", "action", "add", "after", "all", "alter", "analyze", "and", "as", "asc", "attach", "autoincrement", "before", "begin", "between", "by", "cascade", "case", "cast", "check", "collate", "column", "commit", "conflict", "constraint", "create", "cross", "current_date", "current_time", "current_timestamp", "database", "default", "deferrable", "deferred", "delete", "desc", "detach", "distinct", "drop", "each", "else", "end", "escape", "except", "exclusive", "exists", "explain", "fail", "for", "foreign", "from", "full", "glob", "group", "having", "if", "ignore", "immediate", "in", "index", "indexed", "initially", "inner", "insert", "instead", "intersect", "into", "is", "isnull", "join", "key", "left", "like", "limit", "match", "natural", "no", "not", "notnull", "null", "of", "offset", "on", "or", "order", "outer", "plan", "pragma", "primary", "query", "raise", "recursive", "references", "regexp", "reindex", "release", "rename", "replace", "restrict", "right", "rollback", "row", "savepoint", "select", "set", "table", "temp", "temporary", "then", "to", "transaction", "trigger", "union", "unique", "update", "using", "vacuum", "values", "view", "virtual", "when", "where", "with", "without", ], ignore_case=True, ) style = Style.from_dict( { "completion-menu.completion": "bg:#008888 #ffffff", "completion-menu.completion.current": "bg:#00aaaa #000000", "scrollbar.background": "bg:#88aaaa", "scrollbar.button": "bg:#222222", } ) def main(database): connection = sqlite3.connect(database) session = PromptSession( lexer=PygmentsLexer(SqlLexer), completer=sql_completer, style=style ) while True: try: text = session.prompt("> ") except KeyboardInterrupt: continue # Control-C pressed. Try again. except EOFError: break # Control-D pressed. with connection: try: messages = connection.execute(text) except Exception as e: print(repr(e)) else: for message in messages: print(message) print("GoodBye!") if __name__ == "__main__": if len(sys.argv) < 2: db = ":memory:" else: db = sys.argv[1] main(db)