Add filesystem meta-commands and path completion

changelog.rst

@@ -3,6 +3,8 @@ Upcoming
 
 Features:
 ---------
+* Add ``\\cd`` and ``\\ls`` filesystem commands, with path completion for
+  filesystem-aware commands including ``\\e``, ``\\i``, ``\\log-file``, and ``\\o``.
 * Add a ``\service`` prompt token for connections made using ``service=...``
   or ``PGSERVICE``. For example, set ``prompt = '\service> '``.
 

pgcli/main.py

@@ -467,6 +467,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",
@@ -612,6 +624,28 @@ def execute_from_file(self, pattern, **_):
             explain_mode=self.explain_mode,
         )
 
+    def change_directory(self, pattern, **_):
+        try:
+            directory = pathlib.Path(pattern or "~").expanduser()
+            os.chdir(directory)
+        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, **_):
+        try:
+            path = pathlib.Path(pattern or ".").expanduser()
+            path.stat()
+            entries = list(path.iterdir()) if path.is_dir() else [path]
+            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)]
+
     def write_to_logfile(self, pattern, **_):
         if not pattern:
             self.log_file = None

pgcli/packages/sqlcompletion.py

@@ -45,7 +45,11 @@
 Datatype = namedtuple("Datatype", ["schema"])
 Alias = namedtuple("Alias", ["aliases"])
 
-Path = namedtuple("Path", [])
+Path = namedtuple("Path", ["only_directories"])
+Path.__new__.__defaults__ = (False,)
+
+PATH_COMMANDS = frozenset(("\\e", "\\i", "\\log-file", "\\ls", "\\o"))
+DIRECTORY_COMMANDS = frozenset(("\\cd",))
 
 
 class SqlStatement:
@@ -124,8 +128,13 @@ 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 "):
-        return (Path(),)
+    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

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, _, word_before_cursor):
-        completer = PathCompleter(expanduser=True)
-        document = Document(text=word_before_cursor, cursor_position=len(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=path, cursor_position=len(path))
         for c in completer.get_completions(document, None):
             yield Match(completion=c, priority=(0,))
 

tests/test_main.py

@@ -1,4 +1,5 @@
 import os
+import pathlib
 import platform
 import re
 import tempfile
@@ -340,6 +341,58 @@ 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_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()
+    (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)

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",
     [

tests/test_pgspecial.py

@@ -8,6 +8,7 @@
     View,
     Function,
     Datatype,
+    Path,
 )
 
 
@@ -60,6 +61,20 @@ def test_leading_whitespace_ok():
     assert suggestions == suggest_type(cmd, cmd)
 
 
+@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(),)
+
+
+def test_cd_suggests_directories():
+    command = r"\cd "
+    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)