Fix a couple of bugs in the asyncio implementation

Log an informative message if a connection is not closed
and the gc is reclaiming it when using an async dpapi, that
does not support running IO at that stage.

The ``AsyncAdaptedQueue`` used by default on async dpapis
should instantiate a queue only when it's first used
to avoid binding it to a possibly wrong event loop.

Fixes: #5823
Change-Id: Ibfc50e209b1937ae3d6599ae7997f028c7a92c33

4 files changed, 56 insertions, 16 deletions

diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index 829e89b71..aa7e60dfb 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -494,7 +494,6 @@ class AsyncEngine(ProxyComparable, AsyncConnectable):
 
     .. versionadded:: 1.4
 
-
     """  # noqa
 
     # AsyncEngine is a thin proxy; no state should be added here
diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index 47d9e2cba..9b4e61fc3 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -26,7 +26,6 @@ reset_none = util.symbol("reset_none")
 
 
 class _ConnDialect(object):
-
     """partial implementation of :class:`.Dialect`
     which provides DBAPI connection methods.
 
@@ -36,6 +35,8 @@ class _ConnDialect(object):
 
     """
 
+    is_async = False
+
     def do_rollback(self, dbapi_connection):
         dbapi_connection.rollback()
 
@@ -606,11 +607,20 @@ class _ConnectionRecord(object):
 
 
 def _finalize_fairy(
-    connection, connection_record, pool, ref, echo, fairy=None
+    connection,
+    connection_record,
+    pool,
+    ref,  # this is None when called directly, not by the gc
+    echo,
+    fairy=None,
 ):
     """Cleanup for a :class:`._ConnectionFairy` whether or not it's already
     been garbage collected.
 
+    When using an async dialect no IO can happen here (without using
+    a dedicated thread), since this is called outside the greenlet
+    context and with an already running loop. In this case function
+    will only log a message and raise a warning.
     """
 
     if ref:
@@ -624,7 +634,8 @@ def _finalize_fairy(
         assert connection is None
         connection = connection_record.connection
 
-    dont_restore_gced = pool._is_asyncio
+    # null pool is not _is_asyncio but can be used also with async dialects
+    dont_restore_gced = pool._dialect.is_async
 
     if dont_restore_gced:
         detach = not connection_record or ref
@@ -658,11 +669,17 @@ def _finalize_fairy(
 
                     pool._close_connection(connection)
                 else:
-                    util.warn(
-                        "asyncio connection is being garbage "
-                        "collected without being properly closed: %r"
-                        % connection
-                    )
+                    message = (
+                        "The garbage collector is trying to clean up "
+                        "connection %r. This feature is unsupported on async "
+                        "dbapi, since no IO can be performed at this stage to "
+                        "reset the connection. Please close out all "
+                        "connections when they are no longer used, calling "
+                        "``close()`` or using a context manager to "
+                        "manage their lifetime."
+                    ) % connection
+                    pool.logger.error(message)
+                    util.warn(message)
 
         except BaseException as e:
             pool.logger.error(
diff --git a/lib/sqlalchemy/pool/impl.py b/lib/sqlalchemy/pool/impl.py
index 825ac0307..08371a31a 100644
--- a/lib/sqlalchemy/pool/impl.py
+++ b/lib/sqlalchemy/pool/impl.py
@@ -13,6 +13,7 @@
 import traceback
 import weakref
 
+from .base import _ConnDialect
 from .base import _ConnectionFairy
 from .base import _ConnectionRecord
 from .base import Pool
@@ -221,9 +222,14 @@ class QueuePool(Pool):
         return self._pool.maxsize - self._pool.qsize() + self._overflow
 
 
+class _AsyncConnDialect(_ConnDialect):
+    is_async = True
+
+
 class AsyncAdaptedQueuePool(QueuePool):
     _is_asyncio = True
     _queue_class = sqla_queue.AsyncAdaptedQueue
+    _dialect = _AsyncConnDialect()
 
 
 class FallbackAsyncAdaptedQueuePool(AsyncAdaptedQueuePool):
diff --git a/lib/sqlalchemy/util/queue.py b/lib/sqlalchemy/util/queue.py
index ca5a3abde..30e388248 100644
--- a/lib/sqlalchemy/util/queue.py
+++ b/lib/sqlalchemy/util/queue.py
@@ -26,6 +26,7 @@ from .compat import threading
 from .concurrency import asyncio
 from .concurrency import await_fallback
 from .concurrency import await_only
+from .langhelpers import memoized_property
 
 
 __all__ = ["Empty", "Full", "Queue"]
@@ -206,15 +207,32 @@ class AsyncAdaptedQueue:
     await_ = staticmethod(await_only)
 
     def __init__(self, maxsize=0, use_lifo=False):
-        if use_lifo:
-            self._queue = asyncio.LifoQueue(maxsize=maxsize)
-        else:
-            self._queue = asyncio.Queue(maxsize=maxsize)
         self.use_lifo = use_lifo
         self.maxsize = maxsize
-        self.empty = self._queue.empty
-        self.full = self._queue.full
-        self.qsize = self._queue.qsize
+
+    def empty(self):
+        return self._queue.empty()
+
+    def full(self):
+        return self._queue.full()
+
+    def qsize(self):
+        return self._queue.qsize()
+
+    @memoized_property
+    def _queue(self):
+        # Delay creation of the queue until it is first used, to avoid
+        # binding it to a possibly wrong event loop.
+        # By delaying the creation of the pool we accommodate the common
+        # usage pattern of instanciating the engine at module level, where a
+        # different event loop is in present compared to when the application
+        # is actually run.
+
+        if self.use_lifo:
+            queue = asyncio.LifoQueue(maxsize=self.maxsize)
+        else:
+            queue = asyncio.Queue(maxsize=self.maxsize)
+        return queue
 
     def put_nowait(self, item):
         try: