Run handle_error for any exceptions raised in execute_context()

Observing a SQLite connection/cursor being hung on
test_resultset -> PositionalTextTest -> test_dupe_col_obj.
this uses connectionless execution and the result object
fails to be constructed.  When that happens, there is no path
for the cursor or connection to be closed / released.
Recent changes with the exception assertions in #4849 seem to
be causing a cycle to last a little longer than usual which
is exposing this issue for one particular test on SQLite.

As we want to get rid of weakref cleanup, evaluate
why we dont have handle_dbapi_exception for this whole
block, as after_cursor_execute can raise, result construction
can raise, autocommit can raise, close can raise, there
does not seem to be a reason these things should be outside
of the block that gets cleaned up.

Fixes: #5182
Change-Id: I640ac55e8c5f39d287f779fbb5dc0ab727218ca3

doc/build/changelog/unreleased_13/5182.rst

@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, engine
+    :tickets: 5182
+
+    Expanded the scope of cursor/connection cleanup when a statement is
+    executed to include when the result object fails to be constructed, or an
+    after_cursor_execute() event raises an error, or autocommit / autoclose
+    fails.  This allows the DBAPI cursor to be cleaned up on failure and for
+    connectionless execution allows the connection to be closed out and
+    returned to the connection pool, where previously it waiting until garbage
+    collection would trigger a pool return.

lib/sqlalchemy/engine/base.py

@@ -1281,42 +1281,43 @@ def _execute_context(
                     self.dialect.do_execute(
                         cursor, statement, parameters, context
                     )
-        except BaseException as e:
-            self._handle_dbapi_exception(
-                e, statement, parameters, cursor, context
-            )
 
-        if self._has_events or self.engine._has_events:
-            self.dispatch.after_cursor_execute(
-                self,
-                cursor,
-                statement,
-                parameters,
-                context,
-                context.executemany,
-            )
+            if self._has_events or self.engine._has_events:
+                self.dispatch.after_cursor_execute(
+                    self,
+                    cursor,
+                    statement,
+                    parameters,
+                    context,
+                    context.executemany,
+                )
 
-        if context.compiled:
-            context.post_exec()
+            if context.compiled:
+                context.post_exec()
 
-        result = context._setup_result_proxy()
+            result = context._setup_result_proxy()
 
-        if context.should_autocommit and self._root.__transaction is None:
-            self._root._commit_impl(autocommit=True)
+            if context.should_autocommit and self._root.__transaction is None:
+                self._root._commit_impl(autocommit=True)
 
-        # for "connectionless" execution, we have to close this
-        # Connection after the statement is complete.
-        if self.should_close_with_result:
-            assert not context._is_future_result
+            # for "connectionless" execution, we have to close this
+            # Connection after the statement is complete.
+            if self.should_close_with_result:
+                assert not context._is_future_result
+
+                # ResultProxy already exhausted rows / has no rows.
+                # close us now
+                if result._soft_closed:
+                    self.close()
+                else:
+                    # ResultProxy will close this Connection when no more
+                    # rows to fetch.
+                    result._autoclose_connection = True
+        except BaseException as e:
+            self._handle_dbapi_exception(
+                e, statement, parameters, cursor, context
+            )
 
-            # ResultProxy already exhausted rows / has no rows.
-            # close us now
-            if result._soft_closed:
-                self.close()
-            else:
-                # ResultProxy will close this Connection when no more
-                # rows to fetch.
-                result._autoclose_connection = True
         return result
 
     def _cursor_execute(self, cursor, statement, parameters, context=None):

test/engine/test_execute.py

@@ -19,6 +19,7 @@
 from sqlalchemy import Sequence
 from sqlalchemy import String
 from sqlalchemy import testing
+from sqlalchemy import text
 from sqlalchemy import TypeDecorator
 from sqlalchemy import util
 from sqlalchemy import VARCHAR
@@ -2322,6 +2323,74 @@ def handle_error(ctx):
             ):
                 assert_raises(MySpecialException, conn.get_isolation_level)
 
+    @testing.only_on("sqlite")
+    def test_cursor_close_resultset_failed_connectionless(self):
+        engine = engines.testing_engine()
+
+        the_conn = []
+        the_cursor = []
+
+        @event.listens_for(engine, "after_cursor_execute")
+        def go(
+            connection, cursor, statement, parameters, context, executemany
+        ):
+            the_cursor.append(cursor)
+            the_conn.append(connection)
+
+        with mock.patch(
+            "sqlalchemy.engine.result.BaseResult.__init__",
+            Mock(side_effect=tsa.exc.InvalidRequestError("duplicate col")),
+        ):
+            assert_raises(
+                tsa.exc.InvalidRequestError, engine.execute, text("select 1"),
+            )
+
+        # cursor is closed
+        assert_raises_message(
+            engine.dialect.dbapi.ProgrammingError,
+            "Cannot operate on a closed cursor",
+            the_cursor[0].execute,
+            "select 1",
+        )
+
+        # connection is closed
+        assert the_conn[0].closed
+
+    @testing.only_on("sqlite")
+    def test_cursor_close_resultset_failed_explicit(self):
+        engine = engines.testing_engine()
+
+        the_cursor = []
+
+        @event.listens_for(engine, "after_cursor_execute")
+        def go(
+            connection, cursor, statement, parameters, context, executemany
+        ):
+            the_cursor.append(cursor)
+
+        conn = engine.connect()
+
+        with mock.patch(
+            "sqlalchemy.engine.result.BaseResult.__init__",
+            Mock(side_effect=tsa.exc.InvalidRequestError("duplicate col")),
+        ):
+            assert_raises(
+                tsa.exc.InvalidRequestError, conn.execute, text("select 1"),
+            )
+
+        # cursor is closed
+        assert_raises_message(
+            engine.dialect.dbapi.ProgrammingError,
+            "Cannot operate on a closed cursor",
+            the_cursor[0].execute,
+            "select 1",
+        )
+
+        # connection not closed
+        assert not conn.closed
+
+        conn.close()
+
 
 class HandleInvalidatedOnConnectTest(fixtures.TestBase):
     __requires__ = ("sqlite",)