improve error raise for dialect/pool events w/ async engine

Fixed issues where a descriptive error message was not raised for some
classes of event listening with an async engine, which should instead be a
sync engine instance.

Change-Id: I00b9f4fe9373ef5fd5464fac10651cc4024f648e

3 files changed, 26 insertions, 7 deletions

diff --git a/lib/sqlalchemy/engine/events.py b/lib/sqlalchemy/engine/events.py
index 0cbf56a6d..699faf489 100644
--- a/lib/sqlalchemy/engine/events.py
+++ b/lib/sqlalchemy/engine/events.py
@@ -779,7 +779,7 @@ class DialectEvents(event.Events[Dialect]):
     @classmethod
     def _accept_with(
         cls, target: Union[Engine, Type[Engine], Dialect, Type[Dialect]]
-    ) -> Union[Dialect, Type[Dialect]]:
+    ) -> Optional[Union[Dialect, Type[Dialect]]]:
         if isinstance(target, type):
             if issubclass(target, Engine):
                 return Dialect
@@ -787,8 +787,14 @@ class DialectEvents(event.Events[Dialect]):
                 return target
         elif isinstance(target, Engine):
             return target.dialect
-        else:
+        elif isinstance(target, Dialect):
             return target
+        elif hasattr(target, "dispatch") and hasattr(
+            target.dispatch._events, "_no_async_engine_events"
+        ):
+            target.dispatch._events._no_async_engine_events()
+        else:
+            return None
 
     def do_connect(
         self,
diff --git a/lib/sqlalchemy/ext/asyncio/events.py b/lib/sqlalchemy/ext/asyncio/events.py
index a059b93e6..c5d5e0126 100644
--- a/lib/sqlalchemy/ext/asyncio/events.py
+++ b/lib/sqlalchemy/ext/asyncio/events.py
@@ -16,21 +16,29 @@ class AsyncConnectionEvents(engine_event.ConnectionEvents):
     _dispatch_target = AsyncConnectable
 
     @classmethod
-    def _listen(cls, event_key, retval=False):
+    def _no_async_engine_events(cls):
         raise NotImplementedError(
             "asynchronous events are not implemented at this time.  Apply "
             "synchronous listeners to the AsyncEngine.sync_engine or "
             "AsyncConnection.sync_connection attributes."
         )
 
+    @classmethod
+    def _listen(cls, event_key, retval=False):
+        cls._no_async_engine_events()
+
 
 class AsyncSessionEvents(orm_event.SessionEvents):
     _target_class_doc = "SomeSession"
     _dispatch_target = AsyncSession
 
     @classmethod
-    def _listen(cls, event_key, retval=False):
+    def _no_async_engine_events(cls):
         raise NotImplementedError(
             "asynchronous events are not implemented at this time.  Apply "
             "synchronous listeners to the AsyncSession.sync_session."
         )
+
+    @classmethod
+    def _listen(cls, event_key, retval=False):
+        cls._no_async_engine_events()
diff --git a/lib/sqlalchemy/pool/events.py b/lib/sqlalchemy/pool/events.py
index d0d89291b..be2b406a3 100644
--- a/lib/sqlalchemy/pool/events.py
+++ b/lib/sqlalchemy/pool/events.py
@@ -59,7 +59,7 @@ class PoolEvents(event.Events[Pool]):
     @classmethod
     def _accept_with(
         cls, target: Union[Pool, Type[Pool], Engine, Type[Engine]]
-    ) -> Union[Pool, Type[Pool]]:
+    ) -> Optional[Union[Pool, Type[Pool]]]:
         if not typing.TYPE_CHECKING:
             Engine = util.preloaded.engine.Engine
 
@@ -71,9 +71,14 @@ class PoolEvents(event.Events[Pool]):
                 return target
         elif isinstance(target, Engine):
             return target.pool
-        else:
-            assert isinstance(target, Pool)
+        elif isinstance(target, Pool):
             return target
+        elif hasattr(target, "dispatch") and hasattr(
+            target.dispatch._events, "_no_async_engine_events"
+        ):
+            target.dispatch._events._no_async_engine_events()
+        else:
+            return None
 
     @classmethod
     def _listen(  # type: ignore[override]   # would rather keep **kw