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, 27 insertions, 9 deletions

diff --git a/lib/sqlalchemy/event/attr.py b/lib/sqlalchemy/event/attr.py
index 87c6e980f..abb264f98 100644
--- a/lib/sqlalchemy/event/attr.py
+++ b/lib/sqlalchemy/event/attr.py
@@ -41,6 +41,7 @@ from . import registry
 from .. import exc
 from .. import util
 from ..util import threading
+from ..util.concurrency import AsyncAdaptedLock
 
 
 class RefCollection(util.MemoizedSlots):
@@ -277,6 +278,9 @@ class _EmptyListener(_InstanceLevelDispatch):
 class _CompoundListener(_InstanceLevelDispatch):
     __slots__ = "_exec_once_mutex", "_exec_once"
 
+    def _set_asyncio(self):
+        self._exec_once_mutex = AsyncAdaptedLock()
+
     def _memoized_attr__exec_once_mutex(self):
         return threading.Lock()
 
diff --git a/lib/sqlalchemy/event/base.py b/lib/sqlalchemy/event/base.py
index a87c1fe44..c78080738 100644
--- a/lib/sqlalchemy/event/base.py
+++ b/lib/sqlalchemy/event/base.py
@@ -241,8 +241,17 @@ class Events(util.with_metaclass(_EventMeta, object)):
                 return target
 
     @classmethod
-    def _listen(cls, event_key, propagate=False, insert=False, named=False):
-        event_key.base_listen(propagate=propagate, insert=insert, named=named)
+    def _listen(
+        cls,
+        event_key,
+        propagate=False,
+        insert=False,
+        named=False,
+        asyncio=False,
+    ):
+        event_key.base_listen(
+            propagate=propagate, insert=insert, named=named, asyncio=asyncio
+        )
 
     @classmethod
     def _remove(cls, event_key):
diff --git a/lib/sqlalchemy/event/registry.py b/lib/sqlalchemy/event/registry.py
index 19b9174b7..144dd45dc 100644
--- a/lib/sqlalchemy/event/registry.py
+++ b/lib/sqlalchemy/event/registry.py
@@ -244,21 +244,26 @@ class _EventKey(object):
         return self._key in _key_to_collection
 
     def base_listen(
-        self, propagate=False, insert=False, named=False, retval=None
+        self,
+        propagate=False,
+        insert=False,
+        named=False,
+        retval=None,
+        asyncio=False,
     ):
 
         target, identifier = self.dispatch_target, self.identifier
 
         dispatch_collection = getattr(target.dispatch, identifier)
 
+        for_modify = dispatch_collection.for_modify(target.dispatch)
+        if asyncio:
+            for_modify._set_asyncio()
+
         if insert:
-            dispatch_collection.for_modify(target.dispatch).insert(
-                self, propagate
-            )
+            for_modify.insert(self, propagate)
         else:
-            dispatch_collection.for_modify(target.dispatch).append(
-                self, propagate
-            )
+            for_modify.append(self, propagate)
 
     @property
     def _listen_fn(self):