refactor: add and use `Cursor._set_results_from_pipeline()`

This is a cleaner interface between the cursor and the pipeline.

diff --git a/psycopg/psycopg/_pipeline.py b/psycopg/psycopg/_pipeline.py
index b2b32f0bdc65264c0ff3c713b5da93442f0eda3a..b01f269bab97e271b192ef2b9444602317c8b32e 100644 (file)
--- a/psycopg/psycopg/_pipeline.py
+++ b/psycopg/psycopg/_pipeline.py
@@ -114,12 +114,7 @@ class BasePipeline:
                 raise e.OperationalError("pipeline aborted")
         else:
             cursor, prepinfo = queued
-            cursor._check_results(results)
-            if not cursor._results:
-                cursor._results = results
-                cursor._set_current_result(0)
-            else:
-                cursor._results.extend(results)
+            cursor._set_results_from_pipeline(results)
             if prepinfo:
                 key, prep, name = prepinfo
                 # Update the prepare state of the query.

diff --git a/psycopg/psycopg/cursor.py b/psycopg/psycopg/cursor.py
index 40c57241b9a4bfd39100adfc196aaa50be6ce649..d96019d75212e19753940859a224636ea2d47583 100644 (file)
--- a/psycopg/psycopg/cursor.py
+++ b/psycopg/psycopg/cursor.py
@@ -481,6 +481,14 @@ class BaseCursor(Generic[ConnectionType, Row]):
         nrows = self.pgresult.command_tuples
         self._rowcount = nrows if nrows is not None else -1
 
+    def _set_results_from_pipeline(self, results: List["PGresult"]) -> None:
+        self._check_results(results)
+        if not self._results:
+            self._results = results
+            self._set_current_result(0)
+        else:
+            self._results.extend(results)
+
     def _send_prepare(self, name: bytes, query: PostgresQuery) -> None:
         if self._conn._pipeline:
             self._conn._pipeline.command_queue.append(