feat: add ConnectionInfo.pipeline_status

diff --git a/docs/api/objects.rst b/docs/api/objects.rst
index bd3679130c808057cc565875c79c411fb0c500ae..b4e277eeb77ee3272b51b3965e8f2bd0ed144517 100644 (file)
--- a/docs/api/objects.rst
+++ b/docs/api/objects.rst
@@ -34,6 +34,8 @@ Connection information
         is bad. `!ACTIVE` is reported only when a query has been sent to the
         server and not yet completed.
 
+    .. autoattribute:: pipeline_status
+
     .. autoattribute:: backend_pid
     .. autoattribute:: server_version
 

diff --git a/psycopg/psycopg/conninfo.py b/psycopg/psycopg/conninfo.py
index b6ae174969398907d88ee92417a9e3d8545db2aa..767952bca05d14761572dfdf9103b0513113349f 100644 (file)
--- a/psycopg/psycopg/conninfo.py
+++ b/psycopg/psycopg/conninfo.py
@@ -210,6 +210,14 @@ class ConnectionInfo:
         """
         return pq.TransactionStatus(self.pgconn.transaction_status)
 
+    @property
+    def pipeline_status(self) -> pq.PipelineStatus:
+        """
+        The current pipeline status of the client.
+        See :pq:`PQpipelineStatus()`.
+        """
+        return pq.PipelineStatus(self.pgconn.pipeline_status)
+
     def parameter_status(self, param_name: str) -> Optional[str]:
         """
         Return a parameter setting of the connection.

diff --git a/tests/test_conninfo.py b/tests/test_conninfo.py
index 65bc9fc4b1abfe99b52f056b150024b8cd18a4af..a661fc604673e7b9472007ea2610f1f8256cafe9 100644 (file)
--- a/tests/test_conninfo.py
+++ b/tests/test_conninfo.py
@@ -172,6 +172,19 @@ class TestConnectionInfo:
         conn.close()
         assert conn.info.transaction_status.name == "UNKNOWN"
 
+    @pytest.mark.libpq(">= 14")
+    def test_pipeline_status(self, conn):
+        assert not conn.info.pipeline_status
+        assert conn.info.pipeline_status.name == "OFF"
+        with conn.pipeline():
+            assert conn.info.pipeline_status
+            assert conn.info.pipeline_status.name == "ON"
+
+    @pytest.mark.libpq("< 14")
+    def test_pipeline_status_no_pipeline(self, conn):
+        assert not conn.info.pipeline_status
+        assert conn.info.pipeline_status.name == "OFF"
+
     def test_no_password(self, dsn):
         dsn2 = make_conninfo(dsn, password="the-pass-word")
         pgconn = psycopg.pq.PGconn.connect_start(dsn2.encode())