HuaweiCloudDeveloper · 5xuanwen · May 28, 2025 · May 27, 2025 · May 27, 2025 · May 27, 2025
diff --git a/psycopg/psycopg/_tz.py b/psycopg/psycopg/_tz.py
@@ -22,8 +22,8 @@
 
 def get_tzinfo(pgconn: PGconn | None) -> tzinfo:
     """Return the Python timezone info of the connection's timezone."""
-    tzname = pgconn.parameter_status(b"TimeZone") if pgconn else None
     try:
+        tzname = pgconn.exec_(b"SHOW TimeZone").get_value(0, 0) if pgconn else None
         return _timezones[tzname]
     except KeyError:
         sname = tzname.decode() if tzname else "UTC"

diff --git a/psycopg/psycopg/errors.py b/psycopg/psycopg/errors.py
@@ -440,7 +440,7 @@ def severity(self) -> str | None:
 
     @property
     def severity_nonlocalized(self) -> str | None:
-        return self._error_message(DiagnosticField.SEVERITY_NONLOCALIZED)
+        raise NotSupportedError("This is present only in reports generated by libpq versions 9.6 and later.") 
 
     @property
     def sqlstate(self) -> str | None:

diff --git a/psycopg/psycopg/pq/pq_ctypes.py b/psycopg/psycopg/pq/pq_ctypes.py
@@ -82,6 +82,8 @@ class PGconn:
         "_self_ptr",
         "_procpid",
         "__weakref__",
+        "_backend_pid",
+        "_server_version",
     )
 
     def __init__(self, pgconn_ptr: impl.PGconn_struct):
@@ -94,6 +96,8 @@ def __init__(self, pgconn_ptr: impl.PGconn_struct):
         impl.PQsetNoticeReceiver(pgconn_ptr, notice_receiver, byref(self._self_ptr))
 
         self._procpid = getpid()
+        self._backend_pid = None
+        self._server_version = None
 
     def __del__(self) -> None:
         # Close the connection only if it was created in this process,
@@ -131,6 +135,12 @@ def connect_poll(self) -> int:
 
     def finish(self) -> None:
         self._pgconn_ptr, p = None, self._pgconn_ptr
+        self._backend_pid, pid = None, self._backend_pid
+        self._server_version, v = None, self._server_version
+        if pid:
+            del pid
+        if v:
+            del v
         if p:
             PQfinish(p)
 
@@ -236,8 +246,10 @@ def protocol_version(self) -> int:
 
     @property
     def server_version(self) -> str:
-        res = self.exec_(b"select version()")
-        return res.get_value(0, 0).decode().split(" ")[3]
+        if not self._server_version:
+            res = self.exec_(b"select version()")
+            self._server_version = res.get_value(0, 0).decode().split(" ")[3]
+        return self._server_version
 
     @property
     def socket(self) -> int:
@@ -248,7 +260,10 @@ def socket(self) -> int:
 
     @property
     def backend_pid(self) -> int:
-        return self._call_int(impl.PQbackendPID)
+        if not self._backend_pid:
+            res = self.exec_(b"select pg_backend_pid()")
+            self._backend_pid = res.get_value(0, 0).decode()
+        return self._backend_pid
 
     @property
     def needs_password(self) -> bool:

diff --git a/tests/_test_connection.py b/tests/_test_connection.py
@@ -40,7 +40,7 @@ class ParamDef:
     name="isolation_level",
     guc="isolation",
     values=list(psycopg.IsolationLevel),
-    non_default="serializable",
+    non_default="repeatable read",
 )
 param_read_only = ParamDef(
     name="read_only",

diff --git a/tests/test_concurrency_async.py b/tests/test_concurrency_async.py
@@ -125,11 +125,12 @@ async def test_identify_closure(aconn_cls, dsn):
     async def closer():
         await asyncio.sleep(0.2)
         await conn2.execute(
-            "select pg_terminate_backend(%s)", [aconn.pgconn.backend_pid]
+            "select pg_terminate_backend(%s)", [aconn_pid]
         )
 
     aconn = await aconn_cls.connect(dsn)
     conn2 = await aconn_cls.connect(dsn)
+    aconn_pid = aconn.pgconn.backend_pid
     try:
         t = create_task(closer())
         t0 = time.time()

diff --git a/tests/test_connection.py b/tests/test_connection.py
@@ -223,7 +223,7 @@ def test_context_inerror_rollback_no_clobber(conn_cls, conn, dsn, caplog):
         with conn_cls.connect(dsn) as conn2:
             conn2.execute("select 1")
             conn.execute(
-                "select pg_terminate_backend(%s::int)", [conn2.pgconn.backend_pid]
+                "select pg_terminate_backend(%s::bigint)", [conn2.pgconn.backend_pid]
             )
             1 / 0
 
@@ -502,7 +502,7 @@ def cb2(res):
     conn.add_notice_handler(cb1)
     conn.add_notice_handler(cb2)
     conn.add_notice_handler("the wrong thing")
-    conn.add_notice_handler(lambda diag: severities.append(diag.severity_nonlocalized))
+    conn.add_notice_handler(lambda diag: severities.append(diag.severity))
 
     conn.pgconn.exec_(b"set client_min_messages to notice")
     cur = conn.cursor()
@@ -648,6 +648,8 @@ def test_transaction_param_readonly_property(conn, param):
 def test_set_transaction_param_implicit(conn, param, autocommit):
     conn.set_autocommit(autocommit)
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         getattr(conn, f"set_{param.name}")(value)
         cur = conn.execute(
             "select current_setting(%s), current_setting(%s)",
@@ -670,6 +672,8 @@ def test_set_transaction_param_reset(conn, param):
     conn.commit()
 
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         getattr(conn, f"set_{param.name}")(value)
         cur = conn.execute("select current_setting(%s)", [f"transaction_{param.guc}"])
         (pgval,) = cur.fetchone()
@@ -688,6 +692,8 @@ def test_set_transaction_param_reset(conn, param):
 def test_set_transaction_param_block(conn, param, autocommit):
     conn.set_autocommit(autocommit)
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         getattr(conn, f"set_{param.name}")(value)
         with conn.transaction():
             cur = conn.execute(
@@ -881,12 +887,13 @@ def fake_connect_gen(conninfo, **kwargs):
 
 @pytest.mark.crdb_skip("pg_terminate_backend")
 def test_right_exception_on_server_disconnect(conn):
-    with pytest.raises(e.AdminShutdown):
+    with pytest.raises(e.OperationalError):
         conn.execute("select pg_terminate_backend(%s)", [conn.pgconn.backend_pid])
 
 
 @pytest.mark.slow
 @pytest.mark.crdb("skip", reason="error result not returned")
+@pytest.mark.gaussdb_skip("error result not returned")
 def test_right_exception_on_session_timeout(conn):
     want_ex: type[psycopg.Error] = e.IdleInTransactionSessionTimeout
     if sys.platform == "win32":
@@ -895,7 +902,7 @@ def test_right_exception_on_session_timeout(conn):
         # with, not in the client.
         want_ex = psycopg.OperationalError
 
-    conn.execute("SET SESSION idle_in_transaction_session_timeout = 100")
+    conn.execute("SET SESSION idle_in_transaction_timeout = 100")
     sleep(0.2)
     with pytest.raises(want_ex) as ex:
         conn.execute("SELECT * from pg_tables")

diff --git a/tests/test_connection_async.py b/tests/test_connection_async.py
@@ -219,7 +219,7 @@ async def test_context_inerror_rollback_no_clobber(aconn_cls, conn, dsn, caplog)
         async with await aconn_cls.connect(dsn) as conn2:
             await conn2.execute("select 1")
             conn.execute(
-                "select pg_terminate_backend(%s::int)",
+                "select pg_terminate_backend(%s::bigint)",
                 [conn2.pgconn.backend_pid],
             )
             1 / 0
@@ -499,7 +499,7 @@ def cb2(res):
     aconn.add_notice_handler(cb1)
     aconn.add_notice_handler(cb2)
     aconn.add_notice_handler("the wrong thing")
-    aconn.add_notice_handler(lambda diag: severities.append(diag.severity_nonlocalized))
+    aconn.add_notice_handler(lambda diag: severities.append(diag.severity))
 
     aconn.pgconn.exec_(b"set client_min_messages to notice")
     cur = aconn.cursor()
@@ -646,6 +646,8 @@ async def test_transaction_param_readonly_property(aconn, param):
 async def test_set_transaction_param_implicit(aconn, param, autocommit):
     await aconn.set_autocommit(autocommit)
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         await getattr(aconn, f"set_{param.name}")(value)
         cur = await aconn.execute(
             "select current_setting(%s), current_setting(%s)",
@@ -668,6 +670,8 @@ async def test_set_transaction_param_reset(aconn, param):
     await aconn.commit()
 
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         await getattr(aconn, f"set_{param.name}")(value)
         cur = await aconn.execute(
             "select current_setting(%s)", [f"transaction_{param.guc}"]
@@ -690,6 +694,8 @@ async def test_set_transaction_param_reset(aconn, param):
 async def test_set_transaction_param_block(aconn, param, autocommit):
     await aconn.set_autocommit(autocommit)
     for value in param.values:
+        if value == psycopg.IsolationLevel.SERIALIZABLE:
+            pytest.skip("GaussDB currently does not support SERIALIZABLE, which is equivalent to REPEATABLE READ")
         await getattr(aconn, f"set_{param.name}")(value)
         async with aconn.transaction():
             cur = await aconn.execute(
@@ -885,14 +891,15 @@ def fake_connect_gen(conninfo, **kwargs):
 
 @pytest.mark.crdb_skip("pg_terminate_backend")
 async def test_right_exception_on_server_disconnect(aconn):
-    with pytest.raises(e.AdminShutdown):
+    with pytest.raises(e.OperationalError):
         await aconn.execute(
             "select pg_terminate_backend(%s)", [aconn.pgconn.backend_pid]
         )
 
 
 @pytest.mark.slow
 @pytest.mark.crdb("skip", reason="error result not returned")
+@pytest.mark.gaussdb_skip("error result not returned")
 async def test_right_exception_on_session_timeout(aconn):
     want_ex: type[psycopg.Error] = e.IdleInTransactionSessionTimeout
     if sys.platform == "win32":
@@ -901,7 +908,7 @@ async def test_right_exception_on_session_timeout(aconn):
         # with, not in the client.
         want_ex = psycopg.OperationalError
 
-    await aconn.execute("SET SESSION idle_in_transaction_session_timeout = 100")
+    await aconn.execute("SET SESSION idle_in_transaction_timeout = 100")
     await asleep(0.2)
     with pytest.raises(want_ex) as ex:
         await aconn.execute("SELECT * from pg_tables")

diff --git a/tests/test_connection_info.py b/tests/test_connection_info.py
@@ -44,6 +44,7 @@ def test_port(conn):
         conn.info.port
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_get_params(conn, dsn):
     info = conn.info.get_parameters()
     for k, v in conninfo_to_dict(dsn).items():
@@ -53,6 +54,7 @@ def test_get_params(conn, dsn):
             assert k not in info
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_dsn(conn, dsn):
     dsn = conn.info.dsn
     assert "password" not in dsn
@@ -61,6 +63,7 @@ def test_dsn(conn, dsn):
             assert f"{k}=" in dsn
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_get_params_env(conn_cls, dsn, monkeypatch):
     dsn = conninfo_to_dict(dsn)
     dsn.pop("application_name", None)
@@ -74,6 +77,7 @@ def test_get_params_env(conn_cls, dsn, monkeypatch):
         assert conn.info.get_parameters()["application_name"] == "hello test"
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_dsn_env(conn_cls, dsn, monkeypatch):
     dsn = conninfo_to_dict(dsn)
     dsn.pop("application_name", None)
@@ -114,6 +118,7 @@ def test_pipeline_status_no_pipeline(conn):
     assert conn.info.pipeline_status.name == "OFF"
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_no_password(dsn):
     dsn2 = make_conninfo(dsn, password="the-pass-word")
     pgconn = psycopg.pq.PGconn.connect_start(dsn2.encode())
@@ -123,6 +128,7 @@ def test_no_password(dsn):
     assert info.get_parameters()["dbname"] == info.dbname
 
 
+@pytest.mark.gaussdb_skip("This method PGconn.info is not implemented in GaussDB")
 def test_dsn_no_password(dsn):
     dsn2 = make_conninfo(dsn, password="the-pass-word")
     pgconn = psycopg.pq.PGconn.connect_start(dsn2.encode())