perf: drop call to pgconn_encoding per query

Cursor._encoding converted to property. It is only used in error
messages and infrequent operations.

Use uniformly the fallback value of utf-8 in *_encoding() functions,
especially when the connection is closed.

diff --git a/psycopg/psycopg/_encodings.py b/psycopg/psycopg/_encodings.py
index 3401997110322c30fd8aa8e6393be20cfb0a4e4e..c584b26356b6bca12046b9431a6eb2163aa10b56 100644 (file)
--- a/psycopg/psycopg/_encodings.py
+++ b/psycopg/psycopg/_encodings.py
@@ -9,6 +9,7 @@ import string
 import codecs
 from typing import Any, Dict, Optional, TYPE_CHECKING
 
+from .pq._enums import ConnStatus
 from .errors import NotSupportedError
 from ._compat import cache
 
@@ -16,6 +17,9 @@ if TYPE_CHECKING:
     from .pq.abc import PGconn
     from .connection import BaseConnection
 
+OK = ConnStatus.OK
+
+
 _py_codecs = {
     "BIG5": "big5",
     "EUC_CN": "gb2312",
@@ -80,8 +84,9 @@ def conn_encoding(conn: "Optional[BaseConnection[Any]]") -> str:
 
     Default to utf8 if the connection has no encoding info.
     """
-    if not conn:
+    if not conn or conn.closed:
         return "utf-8"
+
     pgenc = conn.pgconn.parameter_status(b"client_encoding") or b"UTF8"
     return pg2pyenc(pgenc)
 
@@ -92,6 +97,9 @@ def pgconn_encoding(pgconn: "PGconn") -> str:
 
     Default to utf8 if the connection has no encoding info.
     """
+    if pgconn.status != OK:
+        return "utf-8"
+
     pgenc = pgconn.parameter_status(b"client_encoding") or b"UTF8"
     return pg2pyenc(pgenc)
 

diff --git a/psycopg/psycopg/cursor.py b/psycopg/psycopg/cursor.py
index 72c128e6f253344f0364df9d546a8133b6273baf..42c3804a1f518a5222c0585b599725727a1a1ca1 100644 (file)
--- a/psycopg/psycopg/cursor.py
+++ b/psycopg/psycopg/cursor.py
@@ -49,7 +49,7 @@ class BaseCursor(Generic[ConnectionType, Row]):
     __slots__ = """
         _conn format _adapters arraysize _closed _results pgresult _pos
         _iresult _rowcount _query _tx _last_query _row_factory _make_row
-        _pgconn _encoding _execmany_returning
+        _pgconn _execmany_returning
         __weakref__
         """.split()
 
@@ -76,7 +76,6 @@ class BaseCursor(Generic[ConnectionType, Row]):
         self._iresult = 0
         self._rowcount = -1
         self._query: Optional[PostgresQuery]
-        self._encoding = "utf-8"
         # None if executemany() not executing, True/False according to returning state
         self._execmany_returning: Optional[bool] = None
         if reset_query:
@@ -383,7 +382,6 @@ class BaseCursor(Generic[ConnectionType, Row]):
             raise e.InterfaceError("the cursor is closed")
 
         self._reset()
-        self._encoding = pgconn_encoding(self._pgconn)
         if not self._last_query or (self._last_query is not query):
             self._last_query = None
             self._tx = adapt.Transformer(self)
@@ -606,7 +604,7 @@ class BaseCursor(Generic[ConnectionType, Row]):
         if status == TUPLES_OK:
             return
         elif status == FATAL_ERROR:
-            raise e.error_from_result(res, encoding=pgconn_encoding(self._pgconn))
+            raise e.error_from_result(res, encoding=self._encoding)
         elif status == PIPELINE_ABORTED:
             raise e.PipelineAborted("pipeline aborted")
         else:
@@ -647,6 +645,10 @@ class BaseCursor(Generic[ConnectionType, Row]):
         self._reset(reset_query=False)
         self._closed = True
 
+    @property
+    def _encoding(self) -> str:
+        return pgconn_encoding(self._pgconn)
+
 
 class Cursor(BaseCursor["Connection[Any]", Row]):
     __module__ = "psycopg"

diff --git a/tests/test_conninfo.py b/tests/test_conninfo.py
index da5c5dbc0ea252932836e78f9501dfe076878e8d..e2c2c01b93c23fb5169413bb77dde2dddaeaefb2 100644 (file)
--- a/tests/test_conninfo.py
+++ b/tests/test_conninfo.py
@@ -226,8 +226,7 @@ class TestConnectionInfo:
         assert ex.value.diag.severity in conn.info.error_message
 
         conn.close()
-        with pytest.raises(psycopg.OperationalError):
-            conn.info.error_message
+        assert "NULL" in conn.info.error_message
 
     @pytest.mark.crdb_skip("backend pid")
     def test_backend_pid(self, conn):