ensure exception raised for all stream w/ sync result

Fixed issue where the :meth:`_asyncio.AsyncSession.execute` method failed
to raise an informative exception if the ``stream_results`` execution
option were used, which is incompatible with a sync-style
:class:`_result.Result` object. An exception is now raised in this scenario
in the same way one is already raised when using ``stream_results`` in
conjunction with the :meth:`_asyncio.AsyncConnection.execute` method.
Additionally, for improved stability with state-sensitive dialects such as
asyncmy, the cursor is now closed when this error condition is raised;
previously with the asyncmy dialect, the connection would go into an
invalid state with unconsumed server side results remaining.

Fixes: #7667
Change-Id: I6eb7affe08584889b57423a90258295f8b7085dc

1 files changed, 22 insertions, 0 deletions

diff --git a/lib/sqlalchemy/ext/asyncio/result.py b/lib/sqlalchemy/ext/asyncio/result.py
index 81ef9915c..62e4a9a0e 100644
--- a/lib/sqlalchemy/ext/asyncio/result.py
+++ b/lib/sqlalchemy/ext/asyncio/result.py
@@ -7,6 +7,7 @@
 
 import operator
 
+from . import exc as async_exc
 from ...engine.result import _NO_ROW
 from ...engine.result import FilterResult
 from ...engine.result import FrozenResult
@@ -646,3 +647,24 @@ class AsyncMappingResult(AsyncCommon):
 
         """
         return await greenlet_spawn(self._only_one_row, True, True, False)
+
+
+async def _ensure_sync_result(result, calling_method):
+    if not result._is_cursor:
+        cursor_result = getattr(result, "raw", None)
+    else:
+        cursor_result = result
+    if cursor_result and cursor_result.context._is_server_side:
+        await greenlet_spawn(cursor_result.close)
+        raise async_exc.AsyncMethodRequired(
+            "Can't use the %s.%s() method with a "
+            "server-side cursor. "
+            "Use the %s.stream() method for an async "
+            "streaming result set."
+            % (
+                calling_method.__self__.__class__.__name__,
+                calling_method.__name__,
+                calling_method.__self__.__class__.__name__,
+            )
+        )
+    return result