mirrors
/
pgcli
mirror of https://github.com/dbcli/pgcli
1
0
Fork 0
Code

Merge branch 'main' into main

This commit is contained in:
Irina Truong 2023-10-06 16:24:24 -07:00 committed by GitHub
parent 6b5b703570 97a1fd6c16
commit 90bd5c7ed8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 261 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
AUTHORS
View File

@ -127,6 +127,8 @@ Contributors:
* Anna Glasgall (annathyst)
* Andy Schoenberger (andyscho)
* Damien Baty (dbaty)
* blag
* Rob Berry (rob-b)
* A23187 (A-23187)
Creator:

5
README.rst
View File

@ -157,8 +157,9 @@ get this running in a development setup.
https://github.com/dbcli/pgcli/blob/master/DEVELOP.rst
Please feel free to reach out to me if you need help.
My email: amjith.r@gmail.com, Twitter: `@amjithr <http://twitter.com/amjithr>`_
Please feel free to reach out to us if you need help.
* Amjith, pgcli author: amjith.r@gmail.com, Twitter: `@amjithr <http://twitter.com/amjithr>`_
* Irina, pgcli maintainer: i.chernyavska@gmail.com, Twitter: `@amjithr <http://twitter.com/irinatruong>`_
Detailed Installation Instructions:
-----------------------------------

8
changelog.rst
View File

@ -1,9 +1,11 @@
========
Upcoming
========
Features:
---------
* Ask for confirmation when quitting cli while a transaction is ongoing.
* New `destructive_statements_require_transaction` config option to refuse to execute a
destructive SQL statement if outside a transaction. This option is off by default.
* Changed the `destructive_warning` config to be a list of commands that are considered
@ -16,6 +18,7 @@ Features:
it will now not restart.
* Config option to always run with a single connection.
* Add comment explaining default LESS environment variable behavior and change example pager setting.
* Added \echo & \qecho special commands. ([issue 1335](https://github.com/dbcli/pgcli/issues/1335)).
Bug fixes:
----------
@ -23,6 +26,11 @@ Bug fixes:
* Fix \ev not producing a correctly quoted "schema"."view"
* Fix 'invalid connection option "dsn"' ([issue 1373](https://github.com/dbcli/pgcli/issues/1373)).
* Fix explain mode when used with `expand`, `auto_expand`, or `--explain-vertical-output` ([issue 1393](https://github.com/dbcli/pgcli/issues/1393)).
* Fix sql-insert format emits NULL as 'None' ([issue 1408](https://github.com/dbcli/pgcli/issues/1408)).
* Improve check for prompt-toolkit 3.0.6 ([issue 1416](https://github.com/dbcli/pgcli/issues/1416)).
* Allow specifying an `alias_map_file` in the config that will use
predetermined table aliases instead of generating aliases programmatically on
the fly
3.5.0 (2022/09/15):
===================

4
pgcli/auth.py
View File

@ -26,7 +26,9 @@ def keyring_initialize(keyring_enabled, *, logger):
try:
keyring = importlib.import_module("keyring")
except ModuleNotFoundError as e:
except (
ModuleNotFoundError
) as e: # ImportError for Python 2, ModuleNotFoundError for Python 3
logger.warning("import keyring failed: %r.", e)

61
pgcli/main.py
View File

@ -279,6 +279,7 @@ class PGCli:
"single_connection": single_connection,
"less_chatty": less_chatty,
"keyword_casing": keyword_casing,
"alias_map_file": c["main"]["alias_map_file"] or None,
}
completer = PGCompleter(
@ -363,6 +364,23 @@ class PGCli:
"Change the table format used to output results",
)
self.pgspecial.register(
self.echo,
"\\echo",
"\\echo [string]",
"Echo a string to stdout",
)
self.pgspecial.register(
self.echo,
"\\qecho",
"\\qecho [string]",
"Echo a string to the query output channel.",
)
def echo(self, pattern, **_):
return [(None, None, None, pattern)]
def change_table_format(self, pattern, **_):
try:
if pattern not in TabularOutputFormatter().supported_formats:
@ -756,7 +774,9 @@ class PGCli:
click.secho(str(e), err=True, fg="red")
else:
try:
if self.output_file and not text.startswith(("\\o ", "\\? ")):
if self.output_file and not text.startswith(
("\\o ", "\\? ", "\\echo ")
):
try:
with open(self.output_file, "a", encoding="utf-8") as f:
click.echo(text, file=f)
@ -800,6 +820,34 @@ class PGCli:
logger.debug("Search path: %r", self.completer.search_path)
return query
def _check_ongoing_transaction_and_allow_quitting(self):
"""Return whether we can really quit, possibly by asking the
user to confirm so if there is an ongoing transaction.
"""
if not self.pgexecute.valid_transaction():
return True
while 1:
try:
choice = click.prompt(
"A transaction is ongoing. Choose `c` to COMMIT, `r` to ROLLBACK, `a` to abort exit.",
default="a",
)
except click.Abort:
# Print newline if user aborts with `^C`, otherwise
# pgcli's prompt will be printed on the same line
# (just after the confirmation prompt).
click.echo(None, err=False)
choice = "a"
choice = choice.lower()
if choice == "a":
return False # do not quit
if choice == "c":
query = self.execute_command("commit")
return query.successful # quit only if query is successful
if choice == "r":
query = self.execute_command("rollback")
return query.successful # quit only if query is successful
def run_cli(self):
logger = self.logger
@ -822,6 +870,10 @@ class PGCli:
text = self.prompt_app.prompt()
except KeyboardInterrupt:
continue
except EOFError:
if not self._check_ongoing_transaction_and_allow_quitting():
continue
raise
try:
text = self.handle_editor_command(text)
@ -831,7 +883,12 @@ class PGCli:
click.secho(str(e), err=True, fg="red")
continue
self.handle_watch_command(text)
try:
self.handle_watch_command(text)
except PgCliQuitError:
if not self._check_ongoing_transaction_and_allow_quitting():
continue
raise
self.now = dt.datetime.today()

9
pgcli/packages/formatter/sqlformatter.py
View File

@ -14,10 +14,13 @@ preprocessors = ()
def escape_for_sql_statement(value):
if value is None:
return "NULL"
if isinstance(value, bytes):
return f"X'{value.hex()}'"
else:
return "'{}'".format(value)
return "'{}'".format(value)
def adapter(data, headers, table_format=None, **kwargs):
@ -29,7 +32,7 @@ def adapter(data, headers, table_format=None, **kwargs):
else:
table_name = table[1]
else:
table_name = '"DUAL"'
table_name = "DUAL"
if table_format == "sql-insert":
h = '", "'.join(headers)
yield 'INSERT INTO "{}" ("{}") VALUES'.format(table_name, h)

8
pgcli/pgclirc
View File

@ -56,6 +56,14 @@ auto_retry_closed_connection = True
# If set to True, table suggestions will include a table alias
generate_aliases = False
# Path to a json file that specifies specific table aliases to use when generate_aliases is set to True
# the format for this file should be:
# {
# "some_table_name": "desired_alias",
# "some_other_table_name": "another_alias"
# }
alias_map_file =
# log_file location.
# In Unix/Linux: ~/.config/pgcli/log
# In Windows: %USERPROFILE%\AppData\Local\dbcli\pgcli\log

28
pgcli/pgcompleter.py
View File

@ -1,3 +1,4 @@
import json
import logging
import re
from itertools import count, repeat, chain
@ -61,18 +62,38 @@ arg_default_type_strip_regex = re.compile(r"::[\w\.]+(\[\])?$")
normalize_ref = lambda ref: ref if ref[0] == '"' else '"' + ref.lower() + '"'
def generate_alias(tbl):
def generate_alias(tbl, alias_map=None):
"""Generate a table alias, consisting of all upper-case letters in
the table name, or, if there are no upper-case letters, the first letter +
all letters preceded by _
param tbl - unescaped name of the table to alias
"""
if alias_map and tbl in alias_map:
return alias_map[tbl]
return "".join(
[l for l in tbl if l.isupper()]
or [l for l, prev in zip(tbl, "_" + tbl) if prev == "_" and l != "_"]
)
class InvalidMapFile(ValueError):
pass
def load_alias_map_file(path):
try:
with open(path) as fo:
alias_map = json.load(fo)
except FileNotFoundError as err:
raise InvalidMapFile(
f"Cannot read alias_map_file - {err.filename} does not exist"
)
except json.JSONDecodeError:
raise InvalidMapFile(f"Cannot read alias_map_file - {path} is not valid json")
else:
return alias_map
class PGCompleter(Completer):
# keywords_tree: A dict mapping keywords to well known following keywords.
# e.g. 'CREATE': ['TABLE', 'USER', ...],
@ -100,6 +121,11 @@ class PGCompleter(Completer):
self.call_arg_oneliner_max = settings.get("call_arg_oneliner_max", 2)
self.search_path_filter = settings.get("search_path_filter")
self.generate_aliases = settings.get("generate_aliases")
alias_map_file = settings.get("alias_map_file")
if alias_map_file is not None:
self.alias_map = load_alias_map_file(alias_map_file)
else:
self.alias_map = None
self.casing_file = settings.get("casing_file")
self.insert_col_skip_patterns = [
re.compile(pattern)

8
pgcli/pgtoolbar.py
View File

@ -1,18 +1,14 @@
from pkg_resources import packaging
import prompt_toolkit
from prompt_toolkit.key_binding.vi_state import InputMode
from prompt_toolkit.application import get_app
parse_version = packaging.version.parse
vi_modes = {
InputMode.INSERT: "I",
InputMode.NAVIGATION: "N",
InputMode.REPLACE: "R",
InputMode.INSERT_MULTIPLE: "M",
}
if parse_version(prompt_toolkit.__version__) >= parse_version("3.0.6"):
# REPLACE_SINGLE is available in prompt_toolkit >= 3.0.6
if "REPLACE_SINGLE" in {e.name for e in InputMode}:
vi_modes[InputMode.REPLACE_SINGLE] = "R"

17
tests/features/basic_commands.feature
View File

@ -23,6 +23,23 @@ Feature: run the cli,
When we send "ctrl + d"
then dbcli exits
Scenario: confirm exit when a transaction is ongoing
When we begin transaction
and we try to send "ctrl + d"
then we see ongoing transaction message
when we send "c"
then dbcli exits
Scenario: cancel exit when a transaction is ongoing
When we begin transaction
and we try to send "ctrl + d"
then we see ongoing transaction message
when we send "a"
then we see dbcli prompt
when we rollback transaction
when we send "ctrl + d"
then dbcli exits
Scenario: interrupt current query via "ctrl + c"
When we send sleep query
and we send "ctrl + c"

34
tests/features/steps/basic_commands.py
View File

@ -64,13 +64,22 @@ def step_ctrl_d(context):
"""
Send Ctrl + D to hopefully exit.
"""
step_try_to_ctrl_d(context)
context.cli.expect(pexpect.EOF, timeout=5)
context.exit_sent = True
@when('we try to send "ctrl + d"')
def step_try_to_ctrl_d(context):
"""
Send Ctrl + D, perhaps exiting, perhaps not (if a transaction is
ongoing).
"""
# turn off pager before exiting
context.cli.sendcontrol("c")
context.cli.sendline(r"\pset pager off")
wrappers.wait_prompt(context)
context.cli.sendcontrol("d")
context.cli.expect(pexpect.EOF, timeout=5)
context.exit_sent = True
@when('we send "ctrl + c"')
@ -87,6 +96,14 @@ def step_see_cancelled_query_warning(context):
wrappers.expect_exact(context, "cancelled query", timeout=2)
@then("we see ongoing transaction message")
def step_see_ongoing_transaction_error(context):
"""
Make sure we receive the warning that a transaction is ongoing.
"""
context.cli.expect("A transaction is ongoing.", timeout=2)
@when("we send sleep query")
def step_send_sleep_15_seconds(context):
"""
@ -199,3 +216,16 @@ def step_resppond_to_destructive_command(context, response):
def step_send_password(context):
wrappers.expect_exact(context, "Password for", timeout=5)
context.cli.sendline(context.conf["pass"] or "DOES NOT MATTER")
@when('we send "{text}"')
def step_send_text(context, text):
context.cli.sendline(text)
# Try to detect whether we are exiting. If so, set `exit_sent`
# so that `after_scenario` correctly cleans up.
try:
context.cli.expect(pexpect.EOF, timeout=0.2)
except pexpect.TIMEOUT:
pass
else:
context.exit_sent = True

4
tests/formatter/test_sqlformatter.py
View File

@ -34,7 +34,7 @@ def test_output_sql_insert():
"Jackson",
"jackson_test@gmail.com",
"132454789",
"",
None,
"2022-09-09 19:44:32.712343+08",
"2022-09-09 19:44:32.712343+08",
]
@ -58,7 +58,7 @@ def test_output_sql_insert():
output_list = [l for l in output]
expected = [
'INSERT INTO "user" ("id", "name", "email", "phone", "description", "created_at", "updated_at") VALUES',
" ('1', 'Jackson', 'jackson_test@gmail.com', '132454789', '', "
" ('1', 'Jackson', 'jackson_test@gmail.com', '132454789', NULL, "
+ "'2022-09-09 19:44:32.712343+08', '2022-09-09 19:44:32.712343+08')",
";",
]

16
tests/test_main.py
View File

@ -296,6 +296,22 @@ def test_i_works(tmpdir, executor):
run(executor, statement, pgspecial=cli.pgspecial)
@dbtest
def test_echo_works(executor):
cli = PGCli(pgexecute=executor)
statement = r"\echo asdf"
result = run(executor, statement, pgspecial=cli.pgspecial)
assert result == ["asdf"]
@dbtest
def test_qecho_works(executor):
cli = PGCli(pgexecute=executor)
statement = r"\qecho asdf"
result = run(executor, statement, pgspecial=cli.pgspecial)
assert result == ["asdf"]
@dbtest
def test_watch_works(executor):
cli = PGCli(pgexecute=executor)

76
tests/test_pgcompleter.py Normal file
View File

@ -0,0 +1,76 @@
import pytest
from pgcli import pgcompleter
def test_load_alias_map_file_missing_file():
with pytest.raises(
pgcompleter.InvalidMapFile,
match=r"Cannot read alias_map_file - /path/to/non-existent/file.json does not exist$",
):
pgcompleter.load_alias_map_file("/path/to/non-existent/file.json")
def test_load_alias_map_file_invalid_json(tmp_path):
fpath = tmp_path / "foo.json"
fpath.write_text("this is not valid json")
with pytest.raises(pgcompleter.InvalidMapFile, match=r".*is not valid json$"):
pgcompleter.load_alias_map_file(str(fpath))
@pytest.mark.parametrize(
"table_name, alias",
[
("SomE_Table", "SET"),
("SOmeTabLe", "SOTL"),
("someTable", "T"),
],
)
def test_generate_alias_uses_upper_case_letters_from_name(table_name, alias):
assert pgcompleter.generate_alias(table_name) == alias
@pytest.mark.parametrize(
"table_name, alias",
[
("some_tab_le", "stl"),
("s_ome_table", "sot"),
("sometable", "s"),
],
)
def test_generate_alias_uses_first_char_and_every_preceded_by_underscore(
table_name, alias
):
assert pgcompleter.generate_alias(table_name) == alias
@pytest.mark.parametrize(
"table_name, alias_map, alias",
[
("some_table", {"some_table": "my_alias"}, "my_alias"),
],
)
def test_generate_alias_can_use_alias_map(table_name, alias_map, alias):
assert pgcompleter.generate_alias(table_name, alias_map) == alias
@pytest.mark.parametrize(
"table_name, alias_map, alias",
[
("SomeTable", {"SomeTable": "my_alias"}, "my_alias"),
],
)
def test_generate_alias_prefers_alias_over_upper_case_name(
table_name, alias_map, alias
):
assert pgcompleter.generate_alias(table_name, alias_map) == alias
@pytest.mark.parametrize(
"table_name, alias",
[
("Some_tablE", "SE"),
("SomeTab_le", "ST"),
],
)
def test_generate_alias_prefers_upper_case_name_over_underscore_name(table_name, alias):
assert pgcompleter.generate_alias(table_name) == alias