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, 3 insertions, 16 deletions

diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index 9bbc04e77..fcf3b974d 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -7,6 +7,7 @@
 from . import exc as async_exc
 from .base import ProxyComparable
 from .base import StartableContext
+from .result import _ensure_sync_result
 from .result import AsyncResult
 from ... import exc
 from ... import inspection
@@ -381,15 +382,8 @@ class AsyncConnection(ProxyComparable, StartableContext, AsyncConnectable):
             execution_options,
             _require_await=True,
         )
-        if result.context._is_server_side:
-            raise async_exc.AsyncMethodRequired(
-                "Can't use the connection.exec_driver_sql() method with a "
-                "server-side cursor."
-                "Use the connection.stream() method for an async "
-                "streaming result set."
-            )
 
-        return result
+        return await _ensure_sync_result(result, self.exec_driver_sql)
 
     async def stream(
         self,
@@ -462,14 +456,7 @@ class AsyncConnection(ProxyComparable, StartableContext, AsyncConnectable):
             execution_options,
             _require_await=True,
         )
-        if result.context._is_server_side:
-            raise async_exc.AsyncMethodRequired(
-                "Can't use the connection.execute() method with a "
-                "server-side cursor."
-                "Use the connection.stream() method for an async "
-                "streaming result set."
-            )
-        return result
+        return await _ensure_sync_result(result, self.execute)
 
     async def scalar(
         self,