handle DBAPI error for fetchall()

Fixed issue in "insertmanyvalues" feature where a particular call to
``cursor.fetchall()`` were not wrapped in SQLAlchemy's exception wrapper,
which apparently can raise a database exception during fetch when using
pyodbc.

Fixes: #11532
Change-Id: Ic07d3e79dd597e18d87a56b45ddffa25e762beb9

diff --git a/doc/build/changelog/unreleased_20/11532.rst b/doc/build/changelog/unreleased_20/11532.rst
new file mode 100644 (file)
index 0000000..141463d
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/11532.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, engine
+    :tickets: 11532
+
+    Fixed issue in "insertmanyvalues" feature where a particular call to
+    ``cursor.fetchall()`` were not wrapped in SQLAlchemy's exception wrapper,
+    which apparently can raise a database exception during fetch when using
+    pyodbc.

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 3451a824476cb74e669743433f574bf6cfc2b58c..3dd3e7b90494707ede619d44cf22c5a83294fd05 100644 (file)
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2029,6 +2029,7 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
         rowcount = 0
 
         for imv_batch in dialect._deliver_insertmanyvalues_batches(
+            self,
             cursor,
             str_statement,
             effective_parameters,
@@ -2049,6 +2050,7 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
                         imv_batch.replaced_parameters,
                         None,
                         context,
+                        is_sub_exec=True,
                     )
 
             sub_stmt = imv_batch.replaced_statement

diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 29bc7ab3eced6a19ac7a99796514eaef472945bd..df4bd41516b4c45bf8c5672c428d06cd37c51094 100644 (file)
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -59,6 +59,7 @@ from ..sql import compiler
 from ..sql import dml
 from ..sql import expression
 from ..sql import type_api
+from ..sql import util as sql_util
 from ..sql._typing import is_tuple_type
 from ..sql.base import _NoArg
 from ..sql.compiler import DDLCompiler
@@ -771,7 +772,13 @@ class DefaultDialect(Dialect):
         connection.execute(expression.ReleaseSavepointClause(name))
 
     def _deliver_insertmanyvalues_batches(
-        self, cursor, statement, parameters, generic_setinputsizes, context
+        self,
+        connection,
+        cursor,
+        statement,
+        parameters,
+        generic_setinputsizes,
+        context,
     ):
         context = cast(DefaultExecutionContext, context)
         compiled = cast(SQLCompiler, context.compiled)
@@ -822,7 +829,17 @@ class DefaultDialect(Dialect):
 
             if is_returning:
 
-                rows = context.fetchall_for_returning(cursor)
+                try:
+                    rows = context.fetchall_for_returning(cursor)
+                except BaseException as be:
+                    connection._handle_dbapi_exception(
+                        be,
+                        sql_util._long_statement(imv_batch.replaced_statement),
+                        imv_batch.replaced_parameters,
+                        None,
+                        context,
+                        is_sub_exec=True,
+                    )
 
                 # I would have thought "is_returning: Final[bool]"
                 # would have assured this but pylance thinks not

diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index d4c5aef7976ab91fc7195082978da7c7b17bb510..40a75975008c932c74cc5eb36675a4cdf9f8c6dd 100644 (file)
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -2147,6 +2147,7 @@ class Dialect(EventTarget):
 
     def _deliver_insertmanyvalues_batches(
         self,
+        connection: Connection,
         cursor: DBAPICursor,
         statement: str,
         parameters: _DBAPIMultiExecuteParams,

diff --git a/lib/sqlalchemy/testing/fixtures/sql.py b/lib/sqlalchemy/testing/fixtures/sql.py
index 830fa2765938b202695c6b00b26b14fdea52e8cd..39e5b08446571e5f758571de5ccdc2f82721d006 100644 (file)
--- a/lib/sqlalchemy/testing/fixtures/sql.py
+++ b/lib/sqlalchemy/testing/fixtures/sql.py
@@ -470,12 +470,22 @@ def insertmanyvalues_fixture(
             return rows
 
     def _deliver_insertmanyvalues_batches(
-        cursor, statement, parameters, generic_setinputsizes, context
+        connection,
+        cursor,
+        statement,
+        parameters,
+        generic_setinputsizes,
+        context,
     ):
         if randomize_rows:
             cursor = RandomCursor(cursor)
         for batch in orig_dialect(
-            cursor, statement, parameters, generic_setinputsizes, context
+            connection,
+            cursor,
+            statement,
+            parameters,
+            generic_setinputsizes,
+            context,
         ):
             if warn_on_downgraded and batch.is_downgraded:
                 util.warn("Batches were downgraded for sorted INSERT")

diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index ebb0b23a5f69642a7bcd9eec8b259b9dd9f7c0a1..f80b4c447ea1f309717528c2740d360c655c75b4 100644 (file)
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -771,6 +771,27 @@ class InsertManyValuesTest(fixtures.RemovesEvents, fixtures.TablesTest):
             Column("x_value", String(50)),
             Column("y_value", String(50)),
         )
+        Table(
+            "uniq_cons",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("data", String(50), unique=True),
+        )
+
+    @testing.variation("use_returning", [True, False])
+    def test_returning_integrity_error(self, connection, use_returning):
+        """test for #11532"""
+
+        stmt = self.tables.uniq_cons.insert()
+        if use_returning:
+            stmt = stmt.returning(self.tables.uniq_cons.c.id)
+
+        # pymssql thought it would be funny to use OperationalError for
+        # a unique key violation.
+        with expect_raises((exc.IntegrityError, exc.OperationalError)):
+            connection.execute(
+                stmt, [{"data": "the data"}, {"data": "the data"}]
+            )
 
     def test_insert_unicode_keys(self, connection):
         table = self.tables["Unitéble2"]