feat: add PipelineAborted exception

It is a condition that likely will have to be managed, so let's not
clobber it with other OperationalError.

diff --git a/psycopg/psycopg/_pipeline.py b/psycopg/psycopg/_pipeline.py
index 2523aaccdeaef02210c27a4f3dac447aa9c253f3..a62aa76b8a9b93909151070ae78331e3d009fb75 100644 (file)
--- a/psycopg/psycopg/_pipeline.py
+++ b/psycopg/psycopg/_pipeline.py
@@ -148,7 +148,7 @@ class BasePipeline:
             if result.status == ExecStatus.FATAL_ERROR:
                 raise e.error_from_result(result, encoding=pgconn_encoding(self.pgconn))
             elif result.status == ExecStatus.PIPELINE_ABORTED:
-                raise e.OperationalError("pipeline aborted")
+                raise e.PipelineAborted("pipeline aborted")
         else:
             cursor, prepinfo = queued
             cursor._set_results_from_pipeline(results)

diff --git a/psycopg/psycopg/cursor.py b/psycopg/psycopg/cursor.py
index ee77166929595135e086cf94f90dc6cb7314cc3c..c32ef0309d8684cbfc4a5682a5da025af6221795 100644 (file)
--- a/psycopg/psycopg/cursor.py
+++ b/psycopg/psycopg/cursor.py
@@ -493,7 +493,7 @@ class BaseCursor(Generic[ConnectionType, Row]):
         if result.status == ExecStatus.FATAL_ERROR:
             raise e.error_from_result(result, encoding=self._encoding)
         elif result.status == ExecStatus.PIPELINE_ABORTED:
-            raise e.OperationalError("pipeline aborted")
+            raise e.PipelineAborted("pipeline aborted")
         elif result.status in self._status_copy:
             raise e.ProgrammingError(
                 "COPY cannot be used with this method; use copy() insead"
@@ -593,7 +593,7 @@ class BaseCursor(Generic[ConnectionType, Row]):
         elif res.status == ExecStatus.FATAL_ERROR:
             raise e.error_from_result(res, encoding=pgconn_encoding(self._pgconn))
         elif res.status == ExecStatus.PIPELINE_ABORTED:
-            raise e.OperationalError("pipeline aborted")
+            raise e.PipelineAborted("pipeline aborted")
         elif res.status != ExecStatus.TUPLES_OK:
             raise e.ProgrammingError("the last operation didn't produce a result")
 

diff --git a/psycopg/psycopg/errors.py b/psycopg/psycopg/errors.py
index 526e78c1d66f0d59117c72f83de209302a8aaac7..b9252c26839166e8b574a7b1eb43994209de1ade 100644 (file)
--- a/psycopg/psycopg/errors.py
+++ b/psycopg/psycopg/errors.py
@@ -199,6 +199,14 @@ class ConnectionTimeout(OperationalError):
     """
 
 
+class PipelineAborted(OperationalError):
+    """
+    Raised when a operation fails because the current pipeline is in aborted state.
+
+    Subclass of `~psycopg.OperationalError`.
+    """
+
+
 class Diagnostic:
     """Details from a database error report."""
 

diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index d87556a3c809d61e3c59c807b7b66d8937ece64b..7dba9ec7b0bd7cefb2f49d6c818a1a5a422f24fd 100644 (file)
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -173,7 +173,7 @@ def test_pipeline_aborted(conn):
         c1 = conn.execute("select 1")
         with pytest.raises(e.UndefinedTable):
             conn.execute("select * from doesnotexist").fetchone()
-        with pytest.raises(e.OperationalError, match="pipeline aborted"):
+        with pytest.raises(e.PipelineAborted):
             conn.execute("select 'aborted'").fetchone()
         # Sync restore the connection in usable state.
         p.sync()

diff --git a/tests/test_pipeline_async.py b/tests/test_pipeline_async.py
index b2a0fd22c3c48dd9e5016b416e0aed17d0054dc6..afee1e6de3a6f23147a537eeeb0ae545c7419035 100644 (file)
--- a/tests/test_pipeline_async.py
+++ b/tests/test_pipeline_async.py
@@ -176,7 +176,7 @@ async def test_pipeline_aborted(aconn):
         c1 = await aconn.execute("select 1")
         with pytest.raises(e.UndefinedTable):
             await (await aconn.execute("select * from doesnotexist")).fetchone()
-        with pytest.raises(e.OperationalError, match="pipeline aborted"):
+        with pytest.raises(e.PipelineAborted):
             await (await aconn.execute("select 'aborted'")).fetchone()
         # Sync restore the connection in usable state.
         await p.sync()