Merge branch 'main' into feature/command-option

Author: diego-feito-stori

.pre-commit-config.yaml

@@ -1,7 +1,7 @@
 repos:
 - repo: https://github.com/astral-sh/ruff-pre-commit
   # Ruff version.
-  rev: v0.11.7
+  rev: v0.14.10
   hooks:
     # Run the linter.
     - id: ruff

AUTHORS

@@ -145,6 +145,8 @@ Contributors:
     * fbdb
     * Charbel Jacquin (charbeljc)
     * Diego
+    * Devadathan M B (devadathanmb)
+    * Charalampos Stratakis
 
 Creator:
 --------

changelog.rst

@@ -1,6 +1,16 @@
 Upcoming (TBD)
 ==============
 
+Features:
+---------
+* Add support for `\\T` prompt escape sequence to display transaction status (similar to psql's `%x`).
+* Add cursor shape support for vi mode. When ``vi = True``, the terminal cursor now
+  reflects the current editing mode: beam in INSERT, block in NORMAL, underline in REPLACE.
+  Uses prompt_toolkit's ``ModalCursorShapeConfig``.
+
+4.4.0 (2025-12-24)
+==================
+
 Features:
 ---------
 * Add support for `init-command` to run when the connection is established.

pgcli/__init__.py

@@ -1 +1 @@
-__version__ = "4.3.0"
+__version__ = "4.4.0"

pgcli/main.py

@@ -48,6 +48,7 @@
 )
 from prompt_toolkit.history import FileHistory
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
+from prompt_toolkit.cursor_shapes import ModalCursorShapeConfig
 from pygments.lexers.sql import PostgresLexer
 
 from pgspecial.main import PGSpecial, NO_QUERY, PAGER_OFF, PAGER_LONG_OUTPUT
@@ -208,6 +209,7 @@ def __init__(
         self.output_file = None
         self.pgspecial = PGSpecial()
 
+        self.hide_named_query_text = "hide_named_query_text" in c["main"] and c["main"].as_bool("hide_named_query_text")
         self.explain_mode = False
         self.multi_line = c["main"].as_bool("multi_line")
         self.multiline_mode = c["main"].get("multi_line_mode", "psql")
@@ -307,7 +309,28 @@ def __init__(
     def quit(self):
         raise PgCliQuitError
 
+    def toggle_named_query_quiet(self):
+        """Toggle hiding of named query text"""
+        self.hide_named_query_text = not self.hide_named_query_text
+        status = "ON" if self.hide_named_query_text else "OFF"
+        message = f"Named query quiet mode: {status}"
+        return [(None, None, None, message)]
+
+    def _is_named_query_execution(self, text):
+        """Check if the command is a named query execution (\n <name>)."""
+        text = text.strip()
+        return text.startswith("\\n ") and not text.startswith("\\ns ") and not text.startswith("\\nd ")
+
     def register_special_commands(self):
+        self.pgspecial.register(
+            self.toggle_named_query_quiet,
+            "\\nq",
+            "\\nq",
+            "Toggle named query quiet mode (hide query text)",
+            arg_type=NO_QUERY,
+            case_sensitive=True,
+        )
+
         self.pgspecial.register(
             self.change_db,
             "\\c",
@@ -828,7 +851,14 @@ def execute_command(self, text, handle_closed_connection=True):
                 if self.output_file and not text.startswith(("\\o ", "\\log-file", "\\? ", "\\echo ")):
                     try:
                         with open(self.output_file, "a", encoding="utf-8") as f:
-                            click.echo(text, file=f)
+                            should_hide = (
+                                self.hide_named_query_text
+                                and query.is_special
+                                and query.successful
+                                and self._is_named_query_execution(text)
+                            )
+                            if not should_hide:
+                                click.echo(text, file=f)
                             click.echo("\n".join(output), file=f)
                             click.echo("", file=f)  # extra newline
                     except OSError as e:
@@ -842,7 +872,14 @@ def execute_command(self, text, handle_closed_connection=True):
                     try:
                         with open(self.log_file, "a", encoding="utf-8") as f:
                             click.echo(dt.datetime.now().isoformat(), file=f)  # timestamp log
-                            click.echo(text, file=f)
+                            should_hide = (
+                                self.hide_named_query_text
+                                and query.is_special
+                                and query.successful
+                                and self._is_named_query_execution(text)
+                            )
+                            if not should_hide:
+                                click.echo(text, file=f)
                             click.echo("\n".join(output), file=f)
                             click.echo("", file=f)  # extra newline
                     except OSError as e:
@@ -1070,6 +1107,7 @@ def get_continuation(width, line_number, is_soft_wrap):
                 enable_suspend=True,
                 editing_mode=EditingMode.VI if self.vi_mode else EditingMode.EMACS,
                 search_ignore_case=True,
+                cursor=ModalCursorShapeConfig(),
             )
 
             return prompt_app
@@ -1155,6 +1193,18 @@ def _evaluate_command(self, text):
                 style_output=self.style_output,
                 max_field_width=self.max_field_width,
             )
+
+            # Hide query text for named queries in quiet mode
+            if (
+                self.hide_named_query_text
+                and is_special
+                and success
+                and self._is_named_query_execution(text)
+                and title
+                and title.startswith("> ")
+            ):
+                title = None
+
             execution = time() - start
             formatted = format_output(title, cur, headers, status, settings, self.explain_mode)
 
@@ -1278,6 +1328,7 @@ def get_prompt(self, string):
         string = string.replace("\\i", str(self.pgexecute.pid) or "(none)")
         string = string.replace("\\#", "#" if self.pgexecute.superuser else ">")
         string = string.replace("\\n", "\n")
+        string = string.replace("\\T", self.pgexecute.transaction_indicator)
         return string
 
     def get_last_query(self):

pgcli/packages/sqlcompletion.py

@@ -270,7 +270,7 @@ def suggest_special(text):
                 return (Schema(), Function(schema=None, usage="special"))
             return (Schema(), rel_type(schema=None))
 
-    if cmd in ["\\n", "\\ns", "\\nd"]:
+    if cmd in ["\\n", "\\ns", "\\nd", "\\nq"]:
         return (NamedQuery(),)
 
     return (Keyword(), Special())

pgcli/pgclirc

@@ -116,6 +116,11 @@ search_path_filter = False
 # Timing of sql statements and table rendering.
 timing = True
 
+# Hide the query text when executing named queries (\n <name>).
+# Only the query results will be displayed.
+# Can be toggled at runtime with \nq command.
+hide_named_query_text = False
+
 # Show/hide the informational toolbar with function keymap at the footer.
 show_bottom_toolbar = True
 
@@ -171,6 +176,7 @@ verbose_errors = False
 # \i - Postgres PID
 # \# - "@" sign if logged in as superuser, '>' in other case
 # \n - Newline
+# \T - Transaction status: '*' if in a valid transaction, '!' if in a failed transaction, '?' if disconnected, empty otherwise
 # \dsn_alias - name of dsn connection string alias if -D option is used (empty otherwise)
 # \x1b[...m - insert ANSI escape sequence
 # eg: prompt = '\x1b[35m\u@\x1b[32m\h:\x1b[36m\d>'

pgcli/pgexecute.py

@@ -298,6 +298,19 @@ def valid_transaction(self):
         status = self.conn.info.transaction_status
         return status == psycopg.pq.TransactionStatus.ACTIVE or status == psycopg.pq.TransactionStatus.INTRANS
 
+    def is_connection_closed(self):
+        return self.conn.info.transaction_status == psycopg.pq.TransactionStatus.UNKNOWN
+
+    @property
+    def transaction_indicator(self):
+        if self.is_connection_closed():
+            return "?"
+        if self.failed_transaction():
+            return "!"
+        if self.valid_transaction():
+            return "*"
+        return ""
+
     def run(
         self,
         statement,
@@ -492,7 +505,8 @@ def view_definition(self, spec):
             else:
                 template = "CREATE OR REPLACE VIEW {name} AS \n{stmt}"
             return (
-                psycopg.sql.SQL(template)
+                psycopg.sql
+                .SQL(template)
                 .format(
                     name=psycopg.sql.Identifier(result.nspname, result.relname),
                     stmt=psycopg.sql.SQL(result.viewdef),

pyproject.toml

@@ -6,7 +6,6 @@ description = "CLI for Postgres Database. With auto-completion and syntax highli
 readme = "README.rst"
 classifiers = [
     "Intended Audience :: Developers",
-    "License :: OSI Approved :: BSD License",
     "Operating System :: Unix",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
@@ -25,7 +24,8 @@ urls = { Homepage = "https://pgcli.com" }
 requires-python = ">=3.9"
 dependencies = [
     "pgspecial>=2.0.0",
-    "click >= 4.1,<8.1.8",
+    # Click 8.1.8 through 8.3.0 have broken pager invocation for multi-argument PAGER values, which causes behave test failures.
+    "click >= 4.1, != 8.1.8, != 8.2.*, != 8.3.0, < 9",
     "Pygments>=2.0",       # Pygments has to be Capitalcased.
     # We still need to use pt-2 unless pt-3 released on Fedora32
     # see: https://github.com/dbcli/pgcli/pull/1197
@@ -85,6 +85,7 @@ pgcli = ["pgclirc", "packages/pgliterals/pgliterals.json"]
 [tool.ruff]
 target-version = 'py39'
 line-length = 140
+show-fixes = true
 
 [tool.ruff.lint]
 select = [

tests/test_main.py

@@ -577,6 +577,47 @@ def test_duration_in_words(duration_in_seconds, words):
     assert duration_in_words(duration_in_seconds) == words
 
 
+@pytest.mark.parametrize(
+    "transaction_indicator,expected",
+    [
+        ("*", "*testuser"),  # valid transaction
+        ("!", "!testuser"),  # failed transaction
+        ("?", "?testuser"),  # connection closed
+        ("", "testuser"),  # idle
+    ],
+)
+def test_get_prompt_with_transaction_status(transaction_indicator, expected):
+    cli = PGCli()
+    cli.pgexecute = mock.MagicMock()
+    cli.pgexecute.user = "testuser"
+    cli.pgexecute.dbname = "testdb"
+    cli.pgexecute.host = "localhost"
+    cli.pgexecute.short_host = "localhost"
+    cli.pgexecute.port = 5432
+    cli.pgexecute.pid = 12345
+    cli.pgexecute.superuser = False
+    cli.pgexecute.transaction_indicator = transaction_indicator
+
+    result = cli.get_prompt("\\T\\u")
+    assert result == expected
+
+
+def test_get_prompt_transaction_status_in_full_prompt():
+    cli = PGCli()
+    cli.pgexecute = mock.MagicMock()
+    cli.pgexecute.user = "user"
+    cli.pgexecute.dbname = "mydb"
+    cli.pgexecute.host = "db.example.com"
+    cli.pgexecute.short_host = "db.example.com"
+    cli.pgexecute.port = 5432
+    cli.pgexecute.pid = 12345
+    cli.pgexecute.superuser = False
+    cli.pgexecute.transaction_indicator = "*"
+
+    result = cli.get_prompt("\\T\\u@\\h:\\d> ")
+    assert result == "*user@db.example.com:mydb> "
+
+
 @dbtest
 def test_notifications(executor):
     run(executor, "listen chan1")