From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Sat, 2 Apr 2022 23:06:17 +0000 (+0200)
Subject: feat: add PipelineAborted exception
X-Git-Tag: 3.1~145^2
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=39faef12151d6121f93a64e1c772160b21a4b912;p=thirdparty%2Fpsycopg.git

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 2523aaccd..a62aa76b8 100644
--- 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 ee7716692..c32ef0309 100644
--- 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 526e78c1d..b9252c268 100644
--- 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 d87556a3c..7dba9ec7b 100644
--- 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 b2a0fd22c..afee1e6de 100644
--- 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()