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

2 files changed, 31 insertions, 8 deletions

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):