Reset Session._flushing when a bulk_* call can't begin its transaction

Fixed bug where a failed ``Session.bulk_insert_mappings()``,
``Session.bulk_update_mappings()`` or ``Session.bulk_save_objects()`` call
could leave the :class:`_orm.Session` permanently in a "flushing" state.

``Session._bulk_save_mappings()`` set ``self._flushing = True`` and called
``self._autobegin_t()._begin()`` before its ``try`` block, so an exception
raised by ``_begin()`` -- such as ``PendingRollbackError`` when the
transaction still needs a rollback from a prior failed flush -- left
``_flushing`` stuck at ``True``.  As neither ``Session.rollback()`` nor
``Session.close()`` reset the flag, a reused Session then raised
"Session is already flushing" for every subsequent flush.

The flag-set and transaction-begin are moved inside a ``try``/``finally``
that wraps the whole method, matching the shape used by ``Session.flush()``,
so that the flag is always cleared regardless of where the failure occurs.

Fixes: #13485
Closes: #13487
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13487
Pull-request-sha: 4b3283906123068744943fa724722ee63b74e007

Change-Id: I5be8b59db500aacf2a76fe31f9c0c854a8324f70

diff --git a/doc/build/changelog/unreleased_20/13485.rst b/doc/build/changelog/unreleased_20/13485.rst
new file mode 100644 (file)
index 0000000..dced037
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/13485.rst
@@ -0,0 +1,15 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 13485
+
+    Fixed bug where a failed :meth:`_orm.Session.bulk_insert_mappings`,
+    :meth:`_orm.Session.bulk_update_mappings` or
+    :meth:`_orm.Session.bulk_save_objects` call could leave the
+    :class:`_orm.Session` permanently in a "flushing" state, such as when the
+    transaction could not be begun because a previous flush had left it
+    needing a rollback.  Unlike :meth:`_orm.Session.flush`, the bulk methods
+    set the internal flushing flag and began the transaction outside of the
+    ``try``/``finally`` block that resets it, so that neither
+    :meth:`_orm.Session.rollback` nor :meth:`_orm.Session.close` would clear
+    it, and every subsequent flush would raise ``InvalidRequestError: Session
+    is already flushing``.  Pull request courtesy Hamody We.

diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 0b3f5785d16873e6c1f53f98031e83b9a76d22f4..ea363708aa17b257eba2eb5d2a375a247afba260 100644 (file)
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -4902,32 +4902,34 @@ class Session(_SessionClassMethods, EventTarget):
         render_nulls: bool,
     ) -> None:
         mapper = _class_to_mapper(mapper)
-        self._flushing = True
 
-        transaction = self._autobegin_t()._begin()
         try:
-            if isupdate:
-                bulk_persistence._bulk_update(
-                    mapper,
-                    mappings,
-                    transaction,
-                    isstates=isstates,
-                    update_changed_only=update_changed_only,
-                )
-            else:
-                bulk_persistence._bulk_insert(
-                    mapper,
-                    mappings,
-                    transaction,
-                    isstates=isstates,
-                    return_defaults=return_defaults,
-                    render_nulls=render_nulls,
-                )
-            transaction.commit()
+            self._flushing = True
 
-        except:
-            with util.safe_reraise():
-                transaction.rollback(_capture_exception=True)
+            transaction = self._autobegin_t()._begin()
+            try:
+                if isupdate:
+                    bulk_persistence._bulk_update(
+                        mapper,
+                        mappings,
+                        transaction,
+                        isstates=isstates,
+                        update_changed_only=update_changed_only,
+                    )
+                else:
+                    bulk_persistence._bulk_insert(
+                        mapper,
+                        mappings,
+                        transaction,
+                        isstates=isstates,
+                        return_defaults=return_defaults,
+                        render_nulls=render_nulls,
+                    )
+                transaction.commit()
+
+            except:
+                with util.safe_reraise():
+                    transaction.rollback(_capture_exception=True)
         finally:
             self._flushing = False
 

diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index da85cef88d945382fd34f61f48a7005564d999ee..3a3c902b6e6ed2ecb71946c618148c3ca2ee8ab8 100644 (file)
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -763,6 +763,40 @@ class SessionStateTest(_fixtures.FixtureTest):
         s4 = maker2(info={"s4": 8})
         eq_(s4.info, {"s4": 8})
 
+    def test_bulk_save_mappings_resets_flushing_on_pending_rollback(self):
+        """a bulk_* call that fails before it can begin its transaction
+        must still reset Session._flushing, the same way flush() does.
+
+        """
+        users, User = self.tables.users, self.classes.User
+        self.mapper_registry.map_imperatively(User, users)
+
+        s = fixture_session()
+        s.add(User(id=1, name="original"))
+        s.commit()
+
+        # leave the transaction needing a rollback, without rolling back
+        s.add(User(id=2, name=None))
+        assert_raises(exc.IntegrityError, s.flush)
+        is_false(s._flushing)
+
+        # PendingRollbackError is raised by _begin(), before the bulk
+        # operation's own try block is reached
+        assert_raises(
+            exc.PendingRollbackError,
+            s.bulk_update_mappings,
+            User,
+            [{"id": 1, "name": "updated"}],
+        )
+        is_false(s._flushing)
+
+        # the Session is usable again, rather than raising
+        # "Session is already flushing" indefinitely
+        s.rollback()
+        s.bulk_update_mappings(User, [{"id": 1, "name": "updated"}])
+        s.commit()
+        eq_(s.get(User, 1).name, "updated")
+
     @testing.variation("session_type", ["plain", "sessionmaker"])
     @testing.variation("merge", [True, False])
     @testing.variation(