Detect non compatible execution in async mode

The SQLAlchemy async mode now detects and raises an informative
error when an non asyncio compatible :term:`DBAPI` is used.
Using a standard ``DBAPI`` with async SQLAlchemy will cause
it to block like any sync call, interrupting the executing asyncio
loop.

Change-Id: I9aed87dc1b0df53e8cb2109495237038aa2cb2d4

5 files changed, 26 insertions, 22 deletions

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 7c1bbb18e..c56c1f020 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -419,8 +419,8 @@ From version 3.24.0 onwards, SQLite supports "upserts" (update or insert)
 of rows into a table via the ``ON CONFLICT`` clause of the ``INSERT``
 statement. A candidate row will only be inserted if that row does not violate
 any unique or primary key constraints. In the case of a unique constraint violation, a
-secondary action can occur which can be either “DO UPDATE”, indicating that
-the data in the target row should be updated, or “DO NOTHING”, which indicates
+secondary action can occur which can be either "DO UPDATE", indicating that
+the data in the target row should be updated, or "DO NOTHING", which indicates
 to silently skip this row.
 
 Conflicts are determined using columns that are part of existing unique
@@ -469,7 +469,7 @@ and :meth:`_sqlite.Insert.on_conflict_do_nothing`:
 Specifying the Target
 ^^^^^^^^^^^^^^^^^^^^^
 
-Both methods supply the “target” of the conflict using column inference:
+Both methods supply the "target" of the conflict using column inference:
 
 * The :paramref:`_sqlite.Insert.on_conflict_do_update.index_elements` argument
   specifies a sequence containing string column names, :class:`_schema.Column`
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