Adapt event exec_once_mutex to asyncio

The pool makes use of a threading.Lock() for the
"first_connect" event.  if the pool is async make sure this
is a greenlet-adapted asyncio lock.

Fixes: #5581
Change-Id: If52415839c7ed82135465f1fe93b95d86c305820

3 files changed, 9 insertions, 0 deletions

diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index f20b63cf5..87383fef7 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -59,6 +59,8 @@ class Pool(log.Identified):
 
     _dialect = _ConnDialect()
 
+    _is_asyncio = False
+
     def __init__(
         self,
         creator,
diff --git a/lib/sqlalchemy/pool/events.py b/lib/sqlalchemy/pool/events.py
index 3954f907f..9443877a9 100644
--- a/lib/sqlalchemy/pool/events.py
+++ b/lib/sqlalchemy/pool/events.py
@@ -54,6 +54,12 @@ class PoolEvents(event.Events):
         else:
             return target
 
+    @classmethod
+    def _listen(cls, event_key, **kw):
+        target = event_key.dispatch_target
+
+        event_key.base_listen(asyncio=target._is_asyncio)
+
     def connect(self, dbapi_connection, connection_record):
         """Called at the moment a particular DBAPI connection is first
         created for a given :class:`_pool.Pool`.
diff --git a/lib/sqlalchemy/pool/impl.py b/lib/sqlalchemy/pool/impl.py
index e1a9f00db..ffdd63671 100644
--- a/lib/sqlalchemy/pool/impl.py
+++ b/lib/sqlalchemy/pool/impl.py
@@ -218,6 +218,7 @@ class QueuePool(Pool):
 
 
 class AsyncAdaptedQueuePool(QueuePool):
+    _is_asyncio = True
     _queue_class = sqla_queue.AsyncAdaptedQueue