From b2bb83f339a631742b976297abc7009c6ba78978 Mon Sep 17 00:00:00 2001 From: Daniel Cavalcante Date: Sat, 8 Aug 2026 09:29:10 -0300 Subject: [PATCH 1/2] Add path completion and filesystem commands --- pgcli/main.py | 33 +++++++++++++++++++++++++++++++++ pgcli/packages/sqlcompletion.py | 8 ++++++-- pgcli/pgcompleter.py | 4 ++-- tests/test_main.py | 32 ++++++++++++++++++++++++++++++++ tests/test_pgspecial.py | 11 +++++++++++ 5 files changed, 84 insertions(+), 4 deletions(-) diff --git a/pgcli/main.py b/pgcli/main.py index 8a9b1b024..243df31d7 100644 --- a/pgcli/main.py +++ b/pgcli/main.py @@ -410,6 +410,18 @@ def refresh_callback(): arg_type=NO_QUERY, ) self.pgspecial.register(self.execute_from_file, "\\i", "\\i filename", "Execute commands from file.") + self.pgspecial.register( + self.change_directory, + "\\cd", + "\\cd [directory]", + "Change the current working directory.", + ) + self.pgspecial.register( + self.list_directory, + "\\ls", + "\\ls [path]", + "List files in a directory.", + ) self.pgspecial.register( self.write_to_file, "\\o", @@ -555,6 +567,27 @@ def execute_from_file(self, pattern, **_): explain_mode=self.explain_mode, ) + def change_directory(self, pattern, **_): + directory = pathlib.Path(pattern or "~").expanduser() + try: + os.chdir(directory) + except OSError as e: + return [(None, None, None, str(e), "", False, True)] + + return [(None, None, None, str(pathlib.Path.cwd()), "", True, True)] + + def list_directory(self, pattern, **_): + path = pathlib.Path(pattern or ".").expanduser() + try: + path.stat() + entries = list(path.iterdir()) if path.is_dir() else [path] + entries.sort(key=lambda entry: (not entry.is_dir(), entry.name.casefold())) + output = "\n".join(entry.name + (os.sep if entry.is_dir() else "") for entry in entries) + except OSError as e: + return [(None, None, None, str(e), "", False, True)] + + return [(None, None, None, output, "", True, True)] + def write_to_logfile(self, pattern, **_): if not pattern: self.log_file = None diff --git a/pgcli/packages/sqlcompletion.py b/pgcli/packages/sqlcompletion.py index 2bb57e3d2..b43ebd4f1 100644 --- a/pgcli/packages/sqlcompletion.py +++ b/pgcli/packages/sqlcompletion.py @@ -45,7 +45,8 @@ Datatype = namedtuple("Datatype", ["schema"]) Alias = namedtuple("Alias", ["aliases"]) -Path = namedtuple("Path", []) +Path = namedtuple("Path", ["only_directories"]) +Path.__new__.__defaults__ = (False,) class SqlStatement: @@ -124,8 +125,11 @@ def suggest_type(full_text, text_before_cursor): A scope for a column category will be a list of tables. """ - if full_text.startswith("\\i "): + command_text = full_text.lstrip() + if command_text.startswith(("\\i ", "\\e ", "\\ls ")): return (Path(),) + if command_text.startswith("\\cd "): + return (Path(only_directories=True),) # This is a temporary hack; the exception handling # here should be removed once sqlparse has been fixed diff --git a/pgcli/pgcompleter.py b/pgcli/pgcompleter.py index 284c6da48..e07f2380c 100644 --- a/pgcli/pgcompleter.py +++ b/pgcli/pgcompleter.py @@ -840,8 +840,8 @@ def get_keyword_matches(self, suggestion, word_before_cursor): return self.find_matches(word_before_cursor, keywords, mode="strict", meta="keyword") - def get_path_matches(self, _, word_before_cursor): - completer = PathCompleter(expanduser=True) + def get_path_matches(self, suggestion, word_before_cursor): + completer = PathCompleter(expanduser=True, only_directories=suggestion.only_directories) document = Document(text=word_before_cursor, cursor_position=len(word_before_cursor)) for c in completer.get_completions(document, None): yield Match(completion=c, priority=(0,)) diff --git a/tests/test_main.py b/tests/test_main.py index 40e96ab05..fa0bf4fa8 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,4 +1,5 @@ import os +import pathlib import platform import re import tempfile @@ -335,6 +336,37 @@ def test_i_works(tmpdir, executor): run(executor, statement, pgspecial=cli.pgspecial) +def test_change_directory(tmp_path, monkeypatch): + cli = object.__new__(PGCli) + target = tmp_path / "target" + target.mkdir() + monkeypatch.chdir(tmp_path) + + result = cli.change_directory(str(target)) + + assert pathlib.Path.cwd() == target + assert result[0][3] == str(target) + + +def test_list_directory(tmp_path): + cli = object.__new__(PGCli) + (tmp_path / "folder").mkdir() + (tmp_path / "query.sql").write_text("select 1") + + result = cli.list_directory(str(tmp_path)) + + assert result[0][3].splitlines() == [f"folder{os.sep}", "query.sql"] + + +def test_list_directory_reports_missing_path(tmp_path): + cli = object.__new__(PGCli) + + result = cli.list_directory(str(tmp_path / "missing")) + + assert result[0][5:] == (False, True) + assert "missing" in result[0][3] + + @dbtest def test_toggle_verbose_errors(executor): cli = PGCli(pgexecute=executor) diff --git a/tests/test_pgspecial.py b/tests/test_pgspecial.py index cd99e321b..f52c1f84a 100644 --- a/tests/test_pgspecial.py +++ b/tests/test_pgspecial.py @@ -8,6 +8,7 @@ View, Function, Datatype, + Path, ) @@ -60,6 +61,16 @@ def test_leading_whitespace_ok(): assert suggestions == suggest_type(cmd, cmd) +@pytest.mark.parametrize("command", [r"\i ", r"\e ", r"\ls "]) +def test_file_commands_suggest_paths(command): + assert suggest_type(command, command) == (Path(),) + + +def test_cd_suggests_directories(): + command = r"\cd " + assert suggest_type(command, command) == (Path(only_directories=True),) + + def test_dT_suggests_schema_or_datatypes(): text = "\\dT " suggestions = suggest_type(text, text) From d55ae2cc9d3bd8ddb98e382afc541309cb5ed169 Mon Sep 17 00:00:00 2001 From: Daniel Cavalcante Date: Wed, 12 Aug 2026 17:26:58 -0300 Subject: [PATCH 2/2] Polish filesystem commands and path completion --- changelog.rst | 2 ++ pgcli/main.py | 13 +++++++------ pgcli/packages/sqlcompletion.py | 15 ++++++++++----- pgcli/pgcompleter.py | 9 ++++++--- tests/test_main.py | 21 +++++++++++++++++++++ tests/test_pgcompleter.py | 27 +++++++++++++++++++++++++++ tests/test_pgspecial.py | 6 +++++- 7 files changed, 78 insertions(+), 15 deletions(-) diff --git a/changelog.rst b/changelog.rst index 0c3905e38..2c3932389 100644 --- a/changelog.rst +++ b/changelog.rst @@ -10,6 +10,8 @@ Bug fixes: Features: --------- +* Add ``\\cd`` and ``\\ls`` filesystem commands, with path completion for + filesystem-aware commands including ``\\e``, ``\\i``, ``\\log-file``, and ``\\o``. * Add ``\\ne `` to edit a named query in the external editor. Loads the named query's SQL into ``$EDITOR``; on save it is written back to the ``[named queries]`` section, creating it if it does not exist. Complements diff --git a/pgcli/main.py b/pgcli/main.py index 243df31d7..f17070f14 100644 --- a/pgcli/main.py +++ b/pgcli/main.py @@ -568,22 +568,23 @@ def execute_from_file(self, pattern, **_): ) def change_directory(self, pattern, **_): - directory = pathlib.Path(pattern or "~").expanduser() try: + directory = pathlib.Path(pattern or "~").expanduser() os.chdir(directory) - except OSError as e: + except (OSError, RuntimeError) as e: return [(None, None, None, str(e), "", False, True)] return [(None, None, None, str(pathlib.Path.cwd()), "", True, True)] def list_directory(self, pattern, **_): - path = pathlib.Path(pattern or ".").expanduser() try: + path = pathlib.Path(pattern or ".").expanduser() path.stat() entries = list(path.iterdir()) if path.is_dir() else [path] - entries.sort(key=lambda entry: (not entry.is_dir(), entry.name.casefold())) - output = "\n".join(entry.name + (os.sep if entry.is_dir() else "") for entry in entries) - except OSError as e: + entries_with_types = [(entry, entry.is_dir()) for entry in entries] + entries_with_types.sort(key=lambda item: (not item[1], item[0].name.casefold(), item[0].name)) + output = "\n".join(entry.name + (os.sep if is_directory else "") for entry, is_directory in entries_with_types) + except (OSError, RuntimeError) as e: return [(None, None, None, str(e), "", False, True)] return [(None, None, None, output, "", True, True)] diff --git a/pgcli/packages/sqlcompletion.py b/pgcli/packages/sqlcompletion.py index b43ebd4f1..0cf495b20 100644 --- a/pgcli/packages/sqlcompletion.py +++ b/pgcli/packages/sqlcompletion.py @@ -48,6 +48,9 @@ Path = namedtuple("Path", ["only_directories"]) Path.__new__.__defaults__ = (False,) +PATH_COMMANDS = frozenset(("\\e", "\\i", "\\log-file", "\\ls", "\\o")) +DIRECTORY_COMMANDS = frozenset(("\\cd",)) + class SqlStatement: def __init__(self, full_text, text_before_cursor): @@ -125,11 +128,13 @@ def suggest_type(full_text, text_before_cursor): A scope for a column category will be a list of tables. """ - command_text = full_text.lstrip() - if command_text.startswith(("\\i ", "\\e ", "\\ls ")): - return (Path(),) - if command_text.startswith("\\cd "): - return (Path(only_directories=True),) + command_text = text_before_cursor.lstrip() + command, _, _ = parse_special_command(command_text) + if command != command_text: + if command in PATH_COMMANDS: + return (Path(),) + if command in DIRECTORY_COMMANDS: + return (Path(only_directories=True),) # This is a temporary hack; the exception handling # here should be removed once sqlparse has been fixed diff --git a/pgcli/pgcompleter.py b/pgcli/pgcompleter.py index e07f2380c..07d68bf1d 100644 --- a/pgcli/pgcompleter.py +++ b/pgcli/pgcompleter.py @@ -5,6 +5,7 @@ import operator from collections import namedtuple, defaultdict, OrderedDict from cli_helpers.tabular_output import TabularOutputFormatter +from pgspecial.main import parse_special_command from pgspecial.namedqueries import NamedQueries from prompt_toolkit.completion import Completer, Completion, PathCompleter from prompt_toolkit.document import Document @@ -484,7 +485,8 @@ def get_completions(self, document, complete_event, smart_completion=None): # Map suggestion type to method # e.g. 'table' -> self.get_table_matches matcher = self.suggestion_matchers[suggestion_type] - matches.extend(matcher(self, suggestion, word_before_cursor)) + completion_text = document.text_before_cursor if suggestion_type is Path else word_before_cursor + matches.extend(matcher(self, suggestion, completion_text)) # Sort matches so highest priorities are first matches = sorted(matches, key=operator.attrgetter("priority"), reverse=True) @@ -840,9 +842,10 @@ def get_keyword_matches(self, suggestion, word_before_cursor): return self.find_matches(word_before_cursor, keywords, mode="strict", meta="keyword") - def get_path_matches(self, suggestion, word_before_cursor): + def get_path_matches(self, suggestion, text_before_cursor): + _, _, path = parse_special_command(text_before_cursor.lstrip()) completer = PathCompleter(expanduser=True, only_directories=suggestion.only_directories) - document = Document(text=word_before_cursor, cursor_position=len(word_before_cursor)) + document = Document(text=path, cursor_position=len(path)) for c in completer.get_completions(document, None): yield Match(completion=c, priority=(0,)) diff --git a/tests/test_main.py b/tests/test_main.py index fa0bf4fa8..9ca0a4861 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -348,6 +348,27 @@ def test_change_directory(tmp_path, monkeypatch): assert result[0][3] == str(target) +def test_change_directory_reports_missing_path(tmp_path, monkeypatch): + cli = object.__new__(PGCli) + monkeypatch.chdir(tmp_path) + + result = cli.change_directory(str(tmp_path / "missing")) + + assert pathlib.Path.cwd() == tmp_path + assert result[0][5:] == (False, True) + assert "missing" in result[0][3] + + +@pytest.mark.parametrize("command", ["change_directory", "list_directory"]) +def test_filesystem_commands_report_expanduser_errors(command): + cli = object.__new__(PGCli) + with mock.patch.object(pathlib.Path, "expanduser", side_effect=RuntimeError("unknown home")): + result = getattr(cli, command)("~missing") + + assert result[0][3] == "unknown home" + assert result[0][5:] == (False, True) + + def test_list_directory(tmp_path): cli = object.__new__(PGCli) (tmp_path / "folder").mkdir() diff --git a/tests/test_pgcompleter.py b/tests/test_pgcompleter.py index 028d02524..bf38db236 100644 --- a/tests/test_pgcompleter.py +++ b/tests/test_pgcompleter.py @@ -2,6 +2,12 @@ import pytest from pgcli import pgcompleter import tempfile +from prompt_toolkit.document import Document + + +def path_completions(completer, text): + document = Document(text=text, cursor_position=len(text)) + return list(completer.get_completions(document, None)) def test_load_alias_map_file_missing_file(): @@ -19,6 +25,27 @@ def test_load_alias_map_file_invalid_json(tmp_path): pgcompleter.load_alias_map_file(str(fpath)) +def test_path_completion_filters_files_for_cd(tmp_path, monkeypatch): + (tmp_path / "folder").mkdir() + (tmp_path / "file.sql").touch() + monkeypatch.chdir(tmp_path) + completer = pgcompleter.PGCompleter() + + assert [completion.text for completion in path_completions(completer, r"\ls f")] == ["ile.sql", "older"] + assert [completion.text for completion in path_completions(completer, r"\cd f")] == ["older"] + + +def test_path_completion_handles_spaces(tmp_path, monkeypatch): + directory = tmp_path / "space directory" + directory.mkdir() + (directory / "query.sql").touch() + monkeypatch.chdir(tmp_path) + + completions = path_completions(pgcompleter.PGCompleter(), r"\i space directory/q") + + assert [completion.text for completion in completions] == ["uery.sql"] + + @pytest.mark.parametrize( "table_name, alias", [ diff --git a/tests/test_pgspecial.py b/tests/test_pgspecial.py index f52c1f84a..322253659 100644 --- a/tests/test_pgspecial.py +++ b/tests/test_pgspecial.py @@ -61,7 +61,7 @@ def test_leading_whitespace_ok(): assert suggestions == suggest_type(cmd, cmd) -@pytest.mark.parametrize("command", [r"\i ", r"\e ", r"\ls "]) +@pytest.mark.parametrize("command", [r"\i ", r"\e ", r"\ls ", r"\o ", r"\log-file "]) def test_file_commands_suggest_paths(command): assert suggest_type(command, command) == (Path(),) @@ -71,6 +71,10 @@ def test_cd_suggests_directories(): assert suggest_type(command, command) == (Path(only_directories=True),) +def test_path_suggestion_uses_text_before_cursor(): + assert suggest_type(r"\ls folder", r"\l") == (Special(),) + + def test_dT_suggests_schema_or_datatypes(): text = "\\dT " suggestions = suggest_type(text, text)