Merge upstream features and bump to v4.4.1

Cherry-pick from upstream:
- Named query quiet mode \nq (dbcli#1551)
- \T prompt transaction status indicator (dbcli#1553)

Also changes -t/--tuples-only to pure boolean flag (fixes bug
where -t consumed next CLI argument as format value).

Made with ❤️ and 🤖 Claude

Author: DiegoDAF

AUTHORS

@@ -144,6 +144,7 @@ Contributors:
     * Jay Knight (jay-knight)
     * fbdb
     * Charbel Jacquin (charbeljc)
+    * Devadathan M B (devadathanmb)
     * Diego
     * Jeronimo Garcia (bechampion)
 

changelog.rst

@@ -36,6 +36,13 @@ Features:
     * New ``host_key_policy`` setting in ``[ssh tunnels]`` config section
     * Options: ``auto-add`` (default, TOFU), ``warn`` (log warning), ``reject`` (known hosts only)
     * ``reject`` mode only connects to hosts already in ``~/.ssh/known_hosts``
+* Add named query quiet mode to hide query text during execution (upstream #1551).
+    * New ``\nq`` command to save queries that execute without printing the query text
+    * ``quiet`` flag stored in named query metadata
+    * Useful for utility queries where only the result matters
+* Add ``\T`` prompt escape sequence to display transaction status (upstream #1553).
+    * Shows transaction state in prompt: empty (idle), ``*`` (in transaction), ``!`` (failed transaction)
+    * Similar to psql's ``%x`` prompt escape
 
 Bug Fixes:
 ----------

pgcli/__init__.py

@@ -1 +1 @@
-__version__ = "4.3.18"
+__version__ = "4.4.1"

pgcli/main.py

@@ -208,6 +208,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")
@@ -342,7 +343,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",
@@ -972,7 +994,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:
@@ -986,7 +1015,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:
@@ -1324,6 +1360,18 @@ def _evaluate_command(self, text):
                 tuples_only=self.tuples_only,
                 show_status=self.show_status,
             )
+
+            # 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)
 
@@ -1447,6 +1495,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

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

pgcli/pgclirc

@@ -134,6 +134,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
 
@@ -192,6 +197,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

@@ -311,6 +311,19 @@ def valid_transaction(self):
         status = self.conn.info.transaction_status  # type: ignore[union-attr]
         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,

tests/test_main.py

@@ -902,6 +902,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")