Merge "Detect non compatible execution in async mode"

4 files changed, 23 insertions, 19 deletions

diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index 7ba2e369b..63c56c34d 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -285,6 +285,15 @@ class NoReferenceError(InvalidRequestError):
     """Raised by ``ForeignKey`` to indicate a reference cannot be resolved."""
 
 
+class AwaitRequired(InvalidRequestError):
+    """Error raised by the async greenlet spawn if no async operation
+    was awaited when it required one
+
+    """
+
+    code = "xd1r"
+
+
 class NoReferencedTableError(NoReferenceError):
     """Raised by ``ForeignKey`` when the referred ``Table`` cannot be
     located.
@@ -355,10 +364,6 @@ class DontWrapMixin(object):
     """
 
 
-# Moved to orm.exc; compatibility definition installed by orm import until 0.6
-UnmappedColumnError = None
-
-
 class StatementError(SQLAlchemyError):
     """An error occurred during execution of a SQL statement.
 
diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index 9e4851dfc..16edcc2b2 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -243,6 +243,7 @@ class AsyncConnection(StartableContext, AsyncConnectable):
             statement,
             parameters,
             execution_options,
+            _require_await=True,
         )
         if result.context._is_server_side:
             raise async_exc.AsyncMethodRequired(
@@ -272,6 +273,7 @@ class AsyncConnection(StartableContext, AsyncConnectable):
             util.EMPTY_DICT.merge_with(
                 execution_options, {"stream_results": True}
             ),
+            _require_await=True,
         )
         if not result.context._is_server_side:
             # TODO: real exception here
@@ -322,6 +324,7 @@ class AsyncConnection(StartableContext, AsyncConnectable):
             statement,
             parameters,
             execution_options,
+            _require_await=True,
         )
         if result.context._is_server_side:
             raise async_exc.AsyncMethodRequired(
diff --git a/lib/sqlalchemy/testing/asyncio.py b/lib/sqlalchemy/testing/asyncio.py
deleted file mode 100644
index 2e274de16..000000000
--- a/lib/sqlalchemy/testing/asyncio.py
+++ /dev/null
@@ -1,14 +0,0 @@
-from .assertions import assert_raises as _assert_raises
-from .assertions import assert_raises_message as _assert_raises_message
-from ..util import await_fallback as await_
-from ..util import greenlet_spawn
-
-
-async def assert_raises_async(except_cls, msg, coroutine):
-    await greenlet_spawn(_assert_raises, except_cls, await_, coroutine)
-
-
-async def assert_raises_message_async(except_cls, msg, coroutine):
-    await greenlet_spawn(
-        _assert_raises_message, except_cls, msg, await_, coroutine
-    )
diff --git a/lib/sqlalchemy/util/_concurrency_py3k.py b/lib/sqlalchemy/util/_concurrency_py3k.py
index dcee05713..8ad3be543 100644
--- a/lib/sqlalchemy/util/_concurrency_py3k.py
+++ b/lib/sqlalchemy/util/_concurrency_py3k.py
@@ -79,7 +79,9 @@ def await_fallback(awaitable: Coroutine) -> Any:
     return current.driver.switch(awaitable)
 
 
-async def greenlet_spawn(fn: Callable, *args, **kwargs) -> Any:
+async def greenlet_spawn(
+    fn: Callable, *args, _require_await=False, **kwargs
+) -> Any:
     """Runs a sync function ``fn`` in a new greenlet.
 
     The sync function can then use :func:`await_` to wait for async
@@ -95,9 +97,11 @@ async def greenlet_spawn(fn: Callable, *args, **kwargs) -> Any:
     # is interrupted by await_, context is not dead and result is a
     # coroutine to wait. If the context is dead the function has
     # returned, and its result can be returned.
+    switch_occurred = False
     try:
         result = context.switch(*args, **kwargs)
         while not context.dead:
+            switch_occurred = True
             try:
                 # wait for a coroutine from await_ and then return its
                 # result back to it.
@@ -112,6 +116,12 @@ async def greenlet_spawn(fn: Callable, *args, **kwargs) -> Any:
     finally:
         # clean up to avoid cycle resolution by gc
         del context.driver
+    if _require_await and not switch_occurred:
+        raise exc.AwaitRequired(
+            "The current operation required an async execution but none was "
+            "detected. This will usually happen when using a non compatible "
+            "DBAPI driver. Please ensure that an async DBAPI is used."
+        )
     return result