Add AdaptedConnection.run_async

Added new method :meth:`.AdaptedConnection.run_async` to the DBAPI
connection interface used by asyncio drivers, which allows methods to be
called against the underlying "driver" connection directly within a
sync-style function where the ``await`` keyword can't be used, such as
within SQLAlchemy event handler functions. The method is analogous to the
:meth:`_asyncio.AsyncConnection.run_sync` method which translates
async-style calls to sync-style. The method is useful for things like
connection-pool on-connect handlers that need to invoke awaitable methods
on the driver connection when it's first created.

Fixes: #7580
Change-Id: I03c98a72bda0234deb19c00095b31a36f19bf36d

1 files changed, 30 insertions, 0 deletions

diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 80c848379..62477f077 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -23,6 +23,7 @@ from typing import Union
 from ..pool import PoolProxiedConnection
 from ..sql.compiler import Compiled  # noqa
 from ..sql.compiler import TypeCompiler  # noqa
+from ..util.concurrency import await_only
 from ..util.typing import _TypeToInstance
 from ..util.typing import NotRequired
 from ..util.typing import Protocol
@@ -2311,5 +2312,34 @@ class AdaptedConnection:
         """The connection object as returned by the driver after a connect."""
         return self._connection
 
+    def run_async(self, fn):
+        """Run the awaitable returned by the given function, which is passed
+        the raw asyncio driver connection.
+
+        This is used to invoke awaitable-only methods on the driver connection
+        within the context of a "synchronous" method, like a connection
+        pool event handler.
+
+        E.g.::
+
+            engine = create_async_engine(...)
+
+            @event.listens_for(engine.sync_engine, "connect")
+            def register_custom_types(dbapi_connection, ...):
+                dbapi_connection.run_async(
+                    lambda connection: connection.set_type_codec(
+                        'MyCustomType', encoder, decoder, ...
+                    )
+                )
+
+        .. versionadded:: 1.4.30
+
+        .. seealso::
+
+            :ref:`asyncio_events_run_async`
+
+        """
+        return await_only(fn(self._connection))
+
     def __repr__(self):
         return "<AdaptedConnection %s>" % self._connection