dbcli/litecli
1
0
Fork 0
Code Issues Activity

Merging upstream version 1.10.0.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-09 17:52:11 +01:00
parent 4551c2c19a
commit 7e40561ca0
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
23 changed files with 298 additions and 99 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
.github/workflows/codeql.yml vendored Normal file
View file

@ -0,0 +1,41 @@
name: "CodeQL"
on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
schedule:
- cron: "13 4 * * 0"
jobs:
analyze:
name: Analyze
runs-on: ubuntu-latest
permissions:
actions: read
contents: read
security-events: write
strategy:
fail-fast: false
matrix:
language: [ python ]
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Initialize CodeQL
uses: github/codeql-action/init@v2
with:
languages: ${{ matrix.language }}
queries: +security-and-quality
- name: Autobuild
uses: github/codeql-action/autobuild@v2
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v2
with:
category: "/language:${{ matrix.language }}"

2
.pre-commit-config.yaml
View file

@ -1,5 +1,5 @@
repos:
- repo: https://github.com/psf/black
rev: 22.3.0
rev: 23.11.0
hooks:
- id: black

25
CHANGELOG.md
View file

@ -1,3 +1,25 @@
## 1.10.0 - 2022-11-19
### Features
* Adding support for startup commands being set in liteclirc and executed on startup. Limited to commands already implemented in litecli. ([[#56](https://github.com/dbcli/litecli/issues/56)])
### Bug Fixes
* Fix [[#146](https://github.com/dbcli/litecli/issues/146)], making sure `.once`
can be used more than once in a session.
* Fixed setting `successful = True` only when query is executed without exceptions so
failing queries get `successful = False` in `query_history`.
* Changed `master` to `main` in CONTRIBUTING.md to reflect GitHubs new default branch
naming.
* Fixed `.once -o <file>` by opening the output file once per statement instead
of for every line of output ([#148](https://github.com/dbcli/litecli/issues/148)).
* Use the sqlite3 API to cancel a running query on interrupt
([#164](https://github.com/dbcli/litecli/issues/164)).
* Skip internal indexes in the .schema output
([#170](https://github.com/dbcli/litecli/issues/170)).
## 1.9.0 - 2022-06-06
### Features
@ -24,7 +46,7 @@
### Bug Fixes
* Upgrade cli_helpers to workaround Pygments regression.
* Use get_terminal_size from shutil instead of click.
* Use get_terminal_size from shutil instead of click.
## 1.7.0 - 2022-01-11
@ -108,3 +130,4 @@
[Shawn Chapla]: https://github.com/shwnchpl
[Zhaolong Zhu]: https://github.com/zzl0
[Zhiming Wang]: https://github.com/zmwangx
[Bjørnar Smestad]: https://brendesmestad.no

4
CONTRIBUTING.md
View file

@ -49,7 +49,7 @@ You'll always get credit for your work.
$ pip install --editable .
```
6. Create a branch for your bugfix or feature based off the `master` branch:
6. Create a branch for your bugfix or feature based off the `main` branch:
```bash
$ git checkout -b <name-of-bugfix-or-feature>
@ -58,7 +58,7 @@ You'll always get credit for your work.
7. While you work on your bugfix or feature, be sure to pull the latest changes from `upstream`. This ensures that your local codebase is up-to-date:
```bash
$ git pull upstream master
$ git pull upstream main
```
8. When your work is ready for the litecli team to review it, push your branch to your fork:

2
litecli/__init__.py
View file

@ -1 +1 @@
__version__ = "1.9.0"
__version__ = "1.10.0"

5
litecli/completion_refresher.py
View file

@ -7,7 +7,6 @@ from .sqlexecute import SQLExecute
class CompletionRefresher(object):
refreshers = OrderedDict()
def __init__(self):
@ -65,7 +64,7 @@ class CompletionRefresher(object):
# if DB is memory, needed to use same connection
executor = sqlexecute
else:
# Create a new sqlexecute method to popoulate the completions.
# Create a new sqlexecute method to populate the completions.
executor = SQLExecute(e.dbname)
# If callbacks is a single function then push it into a list.
@ -79,7 +78,7 @@ class CompletionRefresher(object):
self._restart_refresh.clear()
break
else:
# Break out of while loop if the for loop finishes natually
# Break out of while loop if the for loop finishes naturally
# without hitting the break statement.
break

8
litecli/key_bindings.py
View file

@ -81,4 +81,12 @@ def cli_bindings(cli):
b = event.app.current_buffer
b.complete_state = None
@kb.add("right", filter=completion_is_selected)
def _(event):
"""Accept the completion that is selected in the dropdown menu."""
_logger.debug("Detected right-arrow key.")
b = event.app.current_buffer
b.complete_state = None
return kb

8
litecli/liteclirc
View file

@ -117,6 +117,12 @@ output.header = "#00ff5f bold"
output.odd-row = ""
output.even-row = ""
# Favorite queries.
[favorite_queries]
# Startup commands
# litecli commands or sqlite commands to be executed on startup.
# some of them will require you to have a database attached.
# they will be executed in the same order as they appear in the list.
[startup_commands]
#commands = ".tables", "pragma foreign_keys = ON;"

67
litecli/main.py
View file

@ -59,7 +59,6 @@ PACKAGE_ROOT = os.path.abspath(os.path.dirname(__file__))
class LiteCli(object):
default_prompt = "\\d> "
max_len_prompt = 45
@ -110,6 +109,13 @@ class LiteCli(object):
fg="red",
)
self.logfile = False
# Load startup commands.
try:
self.startup_commands = c["startup_commands"]
except (
KeyError
): # Redundant given the load_config() function that merges in the standard config, but put here to avoid fail if user do not have updated config file.
self.startup_commands = None
self.completion_refresher = CompletionRefresher()
@ -230,7 +236,6 @@ class LiteCli(object):
return [(None, None, None, "Changed prompt format to %s" % arg)]
def initialize_logging(self):
log_file = self.config["main"]["log_file"]
if log_file == "default":
log_file = config_location() + "log"
@ -298,7 +303,6 @@ class LiteCli(object):
return {x: get(x) for x in keys}
def connect(self, database=""):
cnf = {"database": None}
cnf = self.read_my_cnf_files(cnf.keys())
@ -486,29 +490,17 @@ class LiteCli(object):
except EOFError as e:
raise e
except KeyboardInterrupt:
# get last connection id
connection_id_to_kill = sqlexecute.connection_id
logger.debug("connection id to kill: %r", connection_id_to_kill)
# Restart connection to the database
sqlexecute.connect()
try:
for title, cur, headers, status in sqlexecute.run(
"kill %s" % connection_id_to_kill
):
status_str = str(status).lower()
if status_str.find("ok") > -1:
logger.debug(
"cancelled query, connection id: %r, sql: %r",
connection_id_to_kill,
text,
)
self.echo("cancelled query", err=True, fg="red")
sqlexecute.conn.interrupt()
except Exception as e:
self.echo(
"Encountered error while cancelling query: {}".format(e),
err=True,
fg="red",
)
else:
logger.debug("cancelled query")
self.echo("cancelled query", err=True, fg="red")
except NotImplementedError:
self.echo("Not Yet Implemented.", fg="yellow")
except OperationalError as e:
@ -555,7 +547,6 @@ class LiteCli(object):
complete_style = CompleteStyle.READLINE_LIKE
with self._completer_lock:
if self.key_bindings == "vi":
editing_mode = EditingMode.VI
else:
@ -590,6 +581,42 @@ class LiteCli(object):
search_ignore_case=True,
)
def startup_commands():
if self.startup_commands:
if "commands" in self.startup_commands:
for command in self.startup_commands["commands"]:
try:
res = sqlexecute.run(command)
except Exception as e:
click.echo(command)
self.echo(str(e), err=True, fg="red")
else:
click.echo(command)
for title, cur, headers, status in res:
if title == "dot command not implemented":
self.echo(
"The SQLite dot command '"
+ command.split(" ", 1)[0]
+ "' is not yet implemented.",
fg="yellow",
)
else:
output = self.format_output(title, cur, headers)
for line in output:
self.echo(line)
else:
self.echo(
"Could not read commands. The startup commands needs to be formatted as: \n commands = 'command1', 'command2', ...",
fg="yellow",
)
try:
startup_commands()
except Exception as e:
self.echo(
"Could not execute all startup commands: \n" + str(e), fg="yellow"
)
try:
while True:
one_iteration()

2
litecli/packages/completion_engine.py
View file

@ -210,7 +210,7 @@ def suggest_based_on_last_token(token, text_before_cursor, full_text, identifier
# suggest columns that are present in more than one table
return [{"type": "column", "tables": tables, "drop_unique": True}]
elif p.token_first().value.lower() == "select":
# If the lparen is preceeded by a space chances are we're about to
# If the lparen is preceded by a space chances are we're about to
# do a sub-select.
if last_word(text_before_cursor, "all_punctuations").startswith("("):
return [{"type": "keyword"}]

2
litecli/packages/parseutils.py
View file

@ -147,7 +147,7 @@ def extract_table_identifiers(token_stream):
# extract_tables is inspired from examples in the sqlparse lib.
def extract_tables(sql):
"""Extract the table names from an SQL statment.
"""Extract the table names from an SQL statement.
Returns a list of (schema, table, alias) tuples

5
litecli/packages/special/dbcommands.py
View file

@ -69,13 +69,14 @@ def show_schema(cur, arg=None, **_):
args = (arg,)
query = """
SELECT sql FROM sqlite_master
WHERE name==?
WHERE name==? AND sql IS NOT NULL
ORDER BY tbl_name, type DESC, name
"""
else:
args = tuple()
query = """
SELECT sql FROM sqlite_master
WHERE sql IS NOT NULL
ORDER BY tbl_name, type DESC, name
"""
@ -212,7 +213,7 @@ def load_extension(cur, arg, **_):
"Description of a table",
arg_type=PARSED_QUERY,
case_sensitive=True,
aliases=("\\d", "describe", "desc"),
aliases=("\\d", "desc"),
)
def describe(cur, arg, **_):
if arg:

1
litecli/packages/special/favoritequeries.py
View file

@ -3,7 +3,6 @@ from __future__ import unicode_literals
class FavoriteQueries(object):
section_name = "favorite_queries"
usage = """

39
litecli/packages/special/iocommands.py
View file

@ -21,7 +21,7 @@ from litecli.packages.prompt_utils import confirm_destructive_query
use_expanded_output = False
PAGER_ENABLED = True
tee_file = None
once_file = written_to_once_file = None
once_file = once_file_args = written_to_once_file = None
favoritequeries = FavoriteQueries(ConfigObj())
@ -286,7 +286,7 @@ def delete_favorite_query(arg, **_):
return [(None, None, None, status)]
@special_command("system", "system [command]", "Execute a system shell commmand.")
@special_command("system", "system [command]", "Execute a system shell command.")
def execute_system_command(arg, **_):
"""Execute a system shell command."""
usage = "Syntax: system [command].\n"
@ -377,9 +377,9 @@ def write_tee(output):
aliases=("\\o", "\\once"),
)
def set_once(arg, **_):
global once_file
global once_file_args
once_file = parseargfile(arg)
once_file_args = parseargfile(arg)
return [(None, None, None, "")]
@ -387,27 +387,28 @@ def set_once(arg, **_):
@export
def write_once(output):
global once_file, written_to_once_file
if output and once_file:
try:
f = open(**once_file)
except (IOError, OSError) as e:
once_file = None
raise OSError(
"Cannot write to file '{}': {}".format(e.filename, e.strerror)
)
if output and once_file_args:
if once_file is None:
try:
once_file = open(**once_file_args)
except (IOError, OSError) as e:
once_file = None
raise OSError(
"Cannot write to file '{}': {}".format(e.filename, e.strerror)
)
with f:
click.echo(output, file=f, nl=False)
click.echo("\n", file=f, nl=False)
click.echo(output, file=once_file, nl=False)
click.echo("\n", file=once_file, nl=False)
written_to_once_file = True
@export
def unset_once_if_written():
"""Unset the once file, if it has been written to."""
global once_file
if written_to_once_file:
once_file = None
global once_file, once_file_args, written_to_once_file
if once_file and written_to_once_file:
once_file.close()
once_file = once_file_args = written_to_once_file = None
@special_command(
@ -461,7 +462,7 @@ def watch_query(arg, **kwargs):
# Somewhere in the code the pager its activated after every yield,
# so we disable it in every iteration
set_pager_enabled(False)
for (sql, title) in sql_list:
for sql, title in sql_list:
cur.execute(sql)
if cur.description:
headers = [x[0] for x in cur.description]

83
litecli/packages/special/utils.py
View file

@ -46,3 +46,86 @@ def format_uptime(uptime_in_seconds):
uptime = " ".join(uptime_values)
return uptime
def check_if_sqlitedotcommand(command):
"""Does a check if the command supplied is in the list of SQLite dot commands.
:param command: A command (str) supplied from the user
:returns: True/False
"""
sqlite3dotcommands = [
".archive",
".auth",
".backup",
".bail",
".binary",
".cd",
".changes",
".check",
".clone",
".connection",
".databases",
".dbconfig",
".dbinfo",
".dump",
".echo",
".eqp",
".excel",
".exit",
".expert",
".explain",
".filectrl",
".fullschema",
".headers",
".help",
".import",
".imposter",
".indexes",
".limit",
".lint",
".load",
".log",
".mode",
".nonce",
".nullvalue",
".once",
".open",
".output",
".parameter",
".print",
".progress",
".prompt",
".quit",
".read",
".recover",
".restore",
".save",
".scanstats",
".schema",
".selftest",
".separator",
".session",
".sha3sum",
".shell",
".show",
".stats",
".system",
".tables",
".testcase",
".testctrl",
".timeout",
".timer",
".trace",
".vfsinfo",
".vfslist",
".vfsname",
".width",
]
if isinstance(command, str):
command = command.split(" ", 1)[0].lower()
return command in sqlite3dotcommands
else:
return False

1
litecli/sqlcompleter.py
View file

@ -448,7 +448,6 @@ class SQLCompleter(Completer):
suggestions = suggest_type(document.text, document.text_before_cursor)
for suggestion in suggestions:
_logger.debug("Suggestion type: %r", suggestion["type"])
if suggestion["type"] == "column":

29
litecli/sqlexecute.py
View file

@ -1,8 +1,8 @@
import logging
import sqlite3
import uuid
from contextlib import closing
from sqlite3 import OperationalError
from litecli.packages.special.utils import check_if_sqlitedotcommand
import sqlparse
import os.path
@ -18,7 +18,6 @@ _logger = logging.getLogger(__name__)
class SQLExecute(object):
databases_query = """
PRAGMA database_list
"""
@ -51,7 +50,6 @@ class SQLExecute(object):
def __init__(self, database):
self.dbname = database
self._server_type = None
self.connection_id = None
self.conn = None
if not database:
_logger.debug("Database is not specified. Skip connection.")
@ -76,8 +74,6 @@ class SQLExecute(object):
# Update them after the connection is made to ensure that it was a
# successful connection.
self.dbname = db
# retrieve connection id
self.reset_connection_id()
def run(self, statement):
"""Execute the sql in the database and return the results. The results
@ -129,9 +125,12 @@ class SQLExecute(object):
for result in special.execute(cur, sql):
yield result
except special.CommandNotFound: # Regular SQL
_logger.debug("Regular sql statement. sql: %r", sql)
cur.execute(sql)
yield self.get_result(cur)
if check_if_sqlitedotcommand(sql):
yield ("dot command not implemented", None, None, None)
else:
_logger.debug("Regular sql statement. sql: %r", sql)
cur.execute(sql)
yield self.get_result(cur)
def get_result(self, cursor):
"""Get the current result's data from the cursor."""
@ -204,17 +203,3 @@ class SQLExecute(object):
def server_type(self):
self._server_type = ("sqlite3", "3")
return self._server_type
def get_connection_id(self):
if not self.connection_id:
self.reset_connection_id()
return self.connection_id
def reset_connection_id(self):
# Remember current connection id
_logger.debug("Get current connection id")
# res = self.run('select connection_id()')
self.connection_id = uuid.uuid4()
# for title, cur, headers, status in res:
# self.connection_id = cur.fetchone()[0]
_logger.debug("Current connection id: %s", self.connection_id)

3
tasks.py
View file

@ -34,7 +34,7 @@ class BaseCommand(Command, object):
sys.exit(subprocess.call(cmd, shell=shell))
def call_in_sequence(self, cmds, shell=True):
"""Run multiple commmands in a row, exiting if one fails."""
"""Run multiple commands in a row, exiting if one fails."""
for cmd in cmds:
if subprocess.call(cmd, shell=shell) == 1:
sys.exit(1)
@ -75,7 +75,6 @@ class lint(BaseCommand):
class test(TestCommand):
user_options = [("pytest-args=", "a", "Arguments to pass to pytest")]
def initialize_options(self):

7
tests/liteclirc
View file

@ -135,3 +135,10 @@ Token.Toolbar.Arg.Text = nobold
[favorite_queries]
q_param = select * from test where name=?
sh_param = select * from test where id=$1
# Startup commands
# litecli commands or sqlite commands to be executed on startup.
# some of them will require you to have a database attached.
# they will be executed in the same order as they appear in the list.
[startup_commands]
commands = "create table startupcommands(a text)", "insert into startupcommands values('abc')"

14
tests/test_dbspecial.py
View file

@ -1,6 +1,7 @@
from litecli.packages.completion_engine import suggest_type
from test_completion_engine import sorted_dicts
from litecli.packages.special.utils import format_uptime
from litecli.packages.special.utils import check_if_sqlitedotcommand
def test_import_first_argument():
@ -74,3 +75,16 @@ def test_indexes():
{"type": "schema"},
]
)
def test_check_if_sqlitedotcommand():
test_cases = [
[".tables", True],
[".BiNarY", True],
["binary", False],
[234, False],
[".changes test! test", True],
["NotDotcommand", False],
]
for command, expected_result in test_cases:
assert check_if_sqlitedotcommand(command) == expected_result

10
tests/test_main.py
View file

@ -260,3 +260,13 @@ def test_import_command(executor):
"""
assert result.exit_code == 0
assert expected in "".join(result.output)
def test_startup_commands(executor):
m = LiteCli(liteclirc=default_config_file)
assert m.startup_commands["commands"] == [
"create table startupcommands(a text)",
"insert into startupcommands values('abc')",
]
# implement tests on executions of the startupcommands

21
tests/test_smart_completion_public_schema_only.py
View file

@ -15,7 +15,6 @@ metadata = {
@pytest.fixture
def completer():
import litecli.sqlcompleter as sqlcompleter
comp = sqlcompleter.SQLCompleter()
@ -367,17 +366,15 @@ def test_auto_escaped_col_names(completer, complete_event):
Document(text=text, cursor_position=position), complete_event
)
)
assert (
result
== [
Completion(text="*", start_position=0),
Completion(text="`ABC`", start_position=0),
Completion(text="`insert`", start_position=0),
Completion(text="id", start_position=0),
]
+ list(map(Completion, completer.functions))
+ [Completion(text="`select`", start_position=0)]
+ list(map(Completion, sorted(completer.keywords)))
assert result == [
Completion(text="*", start_position=0),
Completion(text="`ABC`", start_position=0),
Completion(text="`insert`", start_position=0),
Completion(text="id", start_position=0),
] + list(map(Completion, completer.functions)) + [
Completion(text="select", start_position=0)
] + list(
map(Completion, sorted(completer.keywords))
)

18
tests/test_sqlexecute.py
View file

@ -94,11 +94,11 @@ def test_invalid_column_name(executor):
@dbtest
def test_unicode_support_in_output(executor):
run(executor, "create table unicodechars(t text)")
run(executor, u"insert into unicodechars (t) values ('é')")
run(executor, "insert into unicodechars (t) values ('é')")
# See issue #24, this raises an exception without proper handling
results = run(executor, u"select * from unicodechars")
assert_result_equal(results, headers=["t"], rows=[(u"é",)])
results = run(executor, "select * from unicodechars")
assert_result_equal(results, headers=["t"], rows=[("é",)])
@dbtest
@ -106,9 +106,9 @@ def test_invalid_unicode_values_dont_choke(executor):
run(executor, "create table unicodechars(t text)")
# \xc3 is not a valid utf-8 char. But we can insert it into the database
# which can break querying if not handled correctly.
run(executor, u"insert into unicodechars (t) values (cast(x'c3' as text))")
run(executor, "insert into unicodechars (t) values (cast(x'c3' as text))")
results = run(executor, u"select * from unicodechars")
results = run(executor, "select * from unicodechars")
assert_result_equal(results, headers=["t"], rows=[("\\xc3",)])
@ -120,13 +120,13 @@ def test_multiple_queries_same_line(executor):
{
"title": None,
"headers": ["'foo'"],
"rows": [(u"foo",)],
"rows": [("foo",)],
"status": "1 row in set",
},
{
"title": None,
"headers": ["'bar'"],
"rows": [(u"bar",)],
"rows": [("bar",)],
"status": "1 row in set",
},
]
@ -369,8 +369,8 @@ def test_cd_command_current_dir(executor):
@dbtest
def test_unicode_support(executor):
results = run(executor, u"SELECT '日本語' AS japanese;")
assert_result_equal(results, headers=["japanese"], rows=[(u"日本語",)])
results = run(executor, "SELECT '日本語' AS japanese;")
assert_result_equal(results, headers=["japanese"], rows=[("日本語",)])
@dbtest