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

1 files changed, 8 insertions, 3 deletions

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