Decode text results defensively when the client encoding cannot decode them (SQL_ASCII)

changelog.rst

@@ -16,6 +16,17 @@ Bug fixes:
   as the OS user. The database argument is now kept, like psql; only when no
   database is given at all does the listing connect to ``postgres``.
 
+* Fix a prompt crash, garbled timezone output and a completion-refresh crash
+  when the client encoding cannot decode text (e.g. SQL_ASCII), where psycopg
+  returns text columns as raw bytes: the socket directory and timezone query
+  results are now decoded defensively (so the prompt no longer raises a
+  ``TypeError`` on Unix socket connections and the timezone startup message no
+  longer shows a ``b'...'``-prefixed value), and the function metadata rows
+  are decoded before building completions (so the background completion
+  refresh no longer dies in ``parse_defaults`` with a ``TypeError``). Same
+  guard as the completion metadata fix for issue #1405; see upstream issues
+  #1484 and #1518.
+
 4.6.0 (2026-08-26)
 ==================
 

pgcli/pgexecute.py

@@ -55,6 +55,23 @@ def register_typecasters(connection):
         connection.adapters.register_loader(forced_text_type, psycopg.types.string.TextLoader)
 
 
+def _decode_if_bytes(value):
+    """psycopg returns text columns as raw bytes when the client encoding
+    cannot be decoded (e.g. SQL_ASCII); decode defensively so callers can
+    treat the value as a regular str. See issues #1484 and #1518."""
+    if isinstance(value, bytes):
+        return value.decode("utf-8", "replace")
+    return value
+
+
+def _decode_row(row):
+    """psycopg returns text columns as raw bytes when the client encoding
+    cannot be decoded (e.g. SQL_ASCII); decode every scalar value and every
+    array element so callers can treat the whole row as regular str values.
+    See issues #1484 and #1518."""
+    return [[_decode_if_bytes(item) for item in value] if isinstance(value, list) else _decode_if_bytes(value) for value in row]
+
+
 # pg3: I don't know what is this
 class ProtocolSafeCursor(psycopg.Cursor):
     """This class wraps and suppresses Protocol Errors with pgbouncer database.
@@ -667,7 +684,7 @@ def get_socket_directory(self):
             _logger.debug("Socket directory Query. sql: %r", self.socket_directory_query)
             cur.execute(self.socket_directory_query)
             result = cur.fetchone()
-            return result[0] if result else ""
+            return _decode_if_bytes(result[0]) if result else ""
 
     def foreignkeys(self):
         """Yields ForeignKey named tuples"""
@@ -797,7 +814,7 @@ def functions(self):
             _logger.debug("Functions Query. sql: %r", query)
             cur.execute(query)
             for row in cur:
-                yield FunctionMetadata(*row)
+                yield FunctionMetadata(*_decode_row(row))
 
     def datatypes(self):
         """Yields tuples of (schema_name, type_name)"""
@@ -899,7 +916,8 @@ def get_timezone(self) -> str:
         query = psycopg.sql.SQL("show time zone")
         with self.conn.cursor() as cur:
             cur.execute(query)
-            return cur.fetchone()[0]
+            result = cur.fetchone()
+            return _decode_if_bytes(result[0]) if result else ""
 
     def set_timezone(self, timezone: str):
         query = psycopg.sql.SQL("set time zone {}").format(psycopg.sql.Identifier(timezone))

tests/test_pgexecute.py

@@ -842,3 +842,94 @@ def test_virtual_database(executor):
     with patch.object(executor, "conn", virtual_connection):
         result = run(executor, "select 1")
         assert "Command not supported" in result
+
+
+# When the client encoding is one psycopg cannot decode (e.g. SQL_ASCII),
+# text columns come back as raw bytes. See issues #1484 and #1518.
+
+
+@dbtest
+def test_get_socket_directory_decodes_sql_ascii_bytes(executor):
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.fetchone.return_value = (b"/var/run/postgresql",)
+        assert executor.get_socket_directory() == "/var/run/postgresql"
+
+
+@dbtest
+def test_get_socket_directory_str_unchanged(executor):
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.fetchone.return_value = ("/var/run/postgresql",)
+        assert executor.get_socket_directory() == "/var/run/postgresql"
+
+
+@dbtest
+def test_get_timezone_decodes_sql_ascii_bytes(executor):
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.fetchone.return_value = (b"America/Argentina/Buenos_Aires",)
+        assert executor.get_timezone() == "America/Argentina/Buenos_Aires"
+
+
+@dbtest
+def test_get_timezone_str_unchanged(executor):
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.fetchone.return_value = ("UTC",)
+        assert executor.get_timezone() == "UTC"
+
+
+@dbtest
+def test_functions_decodes_sql_ascii_bytes(executor):
+    row = (
+        b"public",
+        b"func_with_default",
+        [b"x"],
+        [b"integer"],
+        [b"i"],
+        b"integer",
+        False,
+        False,
+        False,
+        False,
+        b"'10'::integer, NULL",
+    )
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.__iter__.return_value = iter([row])
+        funcs = list(executor.functions())
+
+    assert len(funcs) == 1
+    func = funcs[0]
+    assert func.schema_name == "public"
+    assert func.func_name == "func_with_default"
+    assert func.arg_names == ("x",)
+    assert func.arg_types == ("integer",)
+    assert func.arg_modes == ("i",)
+    assert func.return_type == "integer"
+    assert func.is_public is True
+    assert func.arg_defaults == ("'10'::integer", "NULL")
+
+
+@dbtest
+def test_functions_str_unchanged(executor):
+    row = (
+        "public",
+        "func_plain",
+        ["x"],
+        ["integer"],
+        ["i"],
+        "integer",
+        False,
+        False,
+        False,
+        False,
+        None,
+    )
+    with patch.object(executor.conn, "cursor") as mock_cursor:
+        mock_cursor.return_value.__enter__.return_value.__iter__.return_value = iter([row])
+        funcs = list(executor.functions())
+
+    assert len(funcs) == 1
+    func = funcs[0]
+    assert func.schema_name == "public"
+    assert func.func_name == "func_plain"
+    assert func.arg_names == ("x",)
+    assert func.return_type == "integer"
+    assert func.arg_defaults == ()