Merge "Surface driver connection object when using a proxied dialect"

20 files changed, 323 insertions, 93 deletions

diff --git a/lib/sqlalchemy/connectors/pyodbc.py b/lib/sqlalchemy/connectors/pyodbc.py
index ed7260d6b..c2bbdf7ce 100644
--- a/lib/sqlalchemy/connectors/pyodbc.py
+++ b/lib/sqlalchemy/connectors/pyodbc.py
@@ -183,8 +183,8 @@ class PyODBCConnector(Connector):
         # adjust for ConnectionFairy being present
         # allows attribute set e.g. "connection.autocommit = True"
         # to work properly
-        if hasattr(connection, "connection"):
-            connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            connection = connection.dbapi_connection
 
         if level == "AUTOCOMMIT":
             connection.autocommit = True
diff --git a/lib/sqlalchemy/dialects/mysql/aiomysql.py b/lib/sqlalchemy/dialects/mysql/aiomysql.py
index c9a87145e..c5ba635c2 100644
--- a/lib/sqlalchemy/dialects/mysql/aiomysql.py
+++ b/lib/sqlalchemy/dialects/mysql/aiomysql.py
@@ -13,7 +13,7 @@ r"""
 
 .. warning:: The aiomysql dialect as of September, 2021 appears to be unmaintained
    and no longer functions for Python version 3.10.   Please refer to the
-   :ref:`asyncmy` dialect for current MySQL asyncio functionality.
+   :ref:`asyncmy` dialect for current MySQL/MariaDD asyncio functionality.
 
 The aiomysql dialect is SQLAlchemy's second Python asyncio dialect.
 
@@ -33,6 +33,7 @@ This dialect should normally be used only with the
 from .pymysql import MySQLDialect_pymysql
 from ... import pool
 from ... import util
+from ...engine import AdaptedConnection
 from ...util.concurrency import asyncio
 from ...util.concurrency import await_fallback
 from ...util.concurrency import await_only
@@ -173,7 +174,7 @@ class AsyncAdapt_aiomysql_ss_cursor(AsyncAdapt_aiomysql_cursor):
         return self.await_(self._cursor.fetchall())
 
 
-class AsyncAdapt_aiomysql_connection:
+class AsyncAdapt_aiomysql_connection(AdaptedConnection):
     await_ = staticmethod(await_only)
     __slots__ = ("dbapi", "_connection", "_execute_mutex")
 
@@ -306,5 +307,8 @@ class MySQLDialect_aiomysql(MySQLDialect_pymysql):
 
         return CLIENT.FOUND_ROWS
 
+    def get_driver_connection(self, connection):
+        return connection._connection
+
 
 dialect = MySQLDialect_aiomysql
diff --git a/lib/sqlalchemy/dialects/mysql/asyncmy.py b/lib/sqlalchemy/dialects/mysql/asyncmy.py
index cde43398d..fb96cd686 100644
--- a/lib/sqlalchemy/dialects/mysql/asyncmy.py
+++ b/lib/sqlalchemy/dialects/mysql/asyncmy.py
@@ -31,6 +31,7 @@ This dialect should normally be used only with the
 from .pymysql import MySQLDialect_pymysql
 from ... import pool
 from ... import util
+from ...engine import AdaptedConnection
 from ...util.concurrency import asynccontextmanager
 from ...util.concurrency import asyncio
 from ...util.concurrency import await_fallback
@@ -177,7 +178,7 @@ class AsyncAdapt_asyncmy_ss_cursor(AsyncAdapt_asyncmy_cursor):
         return self.await_(self._cursor.fetchall())
 
 
-class AsyncAdapt_asyncmy_connection:
+class AsyncAdapt_asyncmy_connection(AdaptedConnection):
     await_ = staticmethod(await_only)
     __slots__ = ("dbapi", "_connection", "_execute_mutex", "_ss_cursors")
 
@@ -335,5 +336,8 @@ class MySQLDialect_asyncmy(MySQLDialect_pymysql):
 
         return CLIENT.FOUND_ROWS
 
+    def get_driver_connection(self, connection):
+        return connection._connection
+
 
 dialect = MySQLDialect_asyncmy
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 04b0c1b6d..2bba2f81a 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2687,8 +2687,8 @@ class MySQLDialect(default.DefaultDialect):
         # adjust for ConnectionFairy being present
         # allows attribute set e.g. "connection.autocommit = True"
         # to work properly
-        if hasattr(connection, "connection"):
-            connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            connection = connection.dbapi_connection
 
         self._set_isolation_level(connection, level)
 
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index aab2018bf..3e705dced 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -1085,8 +1085,8 @@ class OracleDialect_cx_oracle(OracleDialect):
         return result
 
     def set_isolation_level(self, connection, level):
-        if hasattr(connection, "connection"):
-            dbapi_connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            dbapi_connection = connection.dbapi_connection
         else:
             dbapi_connection = connection
         if level == "AUTOCOMMIT":
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 825558f26..dc3da224c 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -121,6 +121,7 @@ from ... import exc
 from ... import pool
 from ... import processors
 from ... import util
+from ...engine import AdaptedConnection
 from ...sql import sqltypes
 from ...util.concurrency import asyncio
 from ...util.concurrency import await_fallback
@@ -566,7 +567,7 @@ class AsyncAdapt_asyncpg_ss_cursor(AsyncAdapt_asyncpg_cursor):
         )
 
 
-class AsyncAdapt_asyncpg_connection:
+class AsyncAdapt_asyncpg_connection(AdaptedConnection):
     __slots__ = (
         "dbapi",
         "_connection",
@@ -1045,5 +1046,8 @@ class PGDialect_asyncpg(PGDialect):
 
         return connect
 
+    def get_driver_connection(self, connection):
+        return connection._connection
+
 
 dialect = PGDialect_asyncpg
diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index 3d1051b7d..d42dd9560 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -433,8 +433,8 @@ class PGDialect_pg8000(PGDialect):
         level = level.replace("_", " ")
 
         # adjust for ConnectionFairy possibly being present
-        if hasattr(connection, "connection"):
-            connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            connection = connection.dbapi_connection
 
         if level == "AUTOCOMMIT":
             connection.autocommit = True
@@ -498,8 +498,8 @@ class PGDialect_pg8000(PGDialect):
 
     def set_client_encoding(self, connection, client_encoding):
         # adjust for ConnectionFairy possibly being present
-        if hasattr(connection, "connection"):
-            connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            connection = connection.dbapi_connection
 
         cursor = connection.cursor()
         cursor.execute("SET CLIENT_ENCODING TO '" + client_encoding + "'")
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index c80198825..a5a56cb6b 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -982,8 +982,8 @@ class PGDialect_psycopg2(PGDialect):
     @util.memoized_instancemethod
     def _hstore_oids(self, conn):
         extras = self._psycopg2_extras()
-        if hasattr(conn, "connection"):
-            conn = conn.connection
+        if hasattr(conn, "dbapi_connection"):
+            conn = conn.dbapi_connection
         oids = extras.HstoreAdapter.get_oids(conn)
         if oids is not None and oids[0]:
             return oids[0:2]
diff --git a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
index eb750b0e7..4319e2661 100644
--- a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
@@ -41,6 +41,7 @@ from .base import SQLiteExecutionContext
 from .pysqlite import SQLiteDialect_pysqlite
 from ... import pool
 from ... import util
+from ...engine import AdaptedConnection
 from ...util.concurrency import await_fallback
 from ...util.concurrency import await_only
 
@@ -162,7 +163,7 @@ class AsyncAdapt_aiosqlite_ss_cursor(AsyncAdapt_aiosqlite_cursor):
         return self.await_(self._cursor.fetchall())
 
 
-class AsyncAdapt_aiosqlite_connection:
+class AsyncAdapt_aiosqlite_connection(AdaptedConnection):
     await_ = staticmethod(await_only)
     __slots__ = ("dbapi", "_connection")
 
@@ -328,5 +329,8 @@ class SQLiteDialect_aiosqlite(SQLiteDialect_pysqlite):
 
         return super().is_disconnect(e, connection, cursor)
 
+    def get_driver_connection(self, connection):
+        return connection._connection
+
 
 dialect = SQLiteDialect_aiosqlite
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
index 0f96e8830..e9d5d9682 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
@@ -499,8 +499,8 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
     )
 
     def set_isolation_level(self, connection, level):
-        if hasattr(connection, "connection"):
-            dbapi_connection = connection.connection
+        if hasattr(connection, "dbapi_connection"):
+            dbapi_connection = connection.dbapi_connection
         else:
             dbapi_connection = connection
 
@@ -521,8 +521,8 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
             return re.search(a, b) is not None
 
         def set_regexp(connection):
-            if hasattr(connection, "connection"):
-                dbapi_connection = connection.connection
+            if hasattr(connection, "dbapi_connection"):
+                dbapi_connection = connection.dbapi_connection
             else:
                 dbapi_connection = connection
             dbapi_connection.create_function(
diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index 3761f5005..6306e201d 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -33,6 +33,7 @@ from .cursor import CursorResult
 from .cursor import FullyBufferedResultProxy
 from .cursor import LegacyCursorResult
 from .cursor import ResultProxy
+from .interfaces import AdaptedConnection
 from .interfaces import Compiled
 from .interfaces import Connectable
 from .interfaces import CreateEnginePlugin
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 25ced0343..2444b5c7f 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -440,6 +440,11 @@ class Connection(Connectable):
     def connection(self):
         """The underlying DB-API connection managed by this Connection.
 
+        This is a SQLAlchemy connection-pool proxied connection
+        which then has the attribute
+        :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the
+        actual driver connection.
+
         .. seealso::
 
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 8bd8a121b..eff28e340 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -646,7 +646,7 @@ class DefaultDialect(interfaces.Dialect):
                         % (", ".join(name for name, obj in trans_objs))
                     )
 
-        dbapi_connection = connection.connection.connection
+        dbapi_connection = connection.connection.dbapi_connection
         for name, characteristic, value in characteristic_values:
             characteristic.set_characteristic(self, dbapi_connection, value)
         connection.connection._connection_record.finalize_callback.append(
@@ -779,6 +779,9 @@ class DefaultDialect(interfaces.Dialect):
                 name = unicode(name)  # noqa
         return name
 
+    def get_driver_connection(self, connection):
+        return connection
+
 
 class _RendersLiteral(object):
     def literal_processor(self, dialect):
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 8379c731a..d1484718e 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -1056,7 +1056,21 @@ class Dialect(object):
         .. versionadded:: 1.0.3
 
         """
-        pass
+
+    def get_driver_connection(self, connection):
+        """Returns the connection object as returned by the external driver
+        package.
+
+        For normal dialects that use a DBAPI compliant driver this call
+        will just return the ``connection`` passed as argument.
+        For dialects that instead adapt a non DBAPI compliant driver, like
+        when adapting an asyncio driver, this call will return the
+        connection-like object as returned by the driver.
+
+        .. versionadded:: 1.4.24
+
+        """
+        raise NotImplementedError()
 
 
 class CreateEnginePlugin(object):
@@ -1719,3 +1733,24 @@ class ExceptionContext(object):
     .. versionadded:: 1.0.3
 
     """
+
+
+class AdaptedConnection(object):
+    """Interface of an adapted connection object to support the DBAPI protocol.
+
+    Used by asyncio dialects to provide a sync-style pep-249 facade on top
+    of the asyncio connection/cursor API provided by the driver.
+
+    .. versionadded:: 1.4.24
+
+    """
+
+    __slots__ = ("_connection",)
+
+    @property
+    def driver_connection(self):
+        """The connection object as returned by the driver after a connect."""
+        return self._connection
+
+    def __repr__(self):
+        return "<AdaptedConnection %s>" % self._connection
diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index ab29438ed..a9e43a65f 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -113,7 +113,6 @@ class AsyncConnection(ProxyComparable, StartableContext, AsyncConnectable):
     def connection(self):
         """Not implemented for async; call
         :meth:`_asyncio.AsyncConnection.get_raw_connection`.
-
         """
         raise exc.InvalidRequestError(
             "AsyncConnection.connection accessor is not implemented as the "
@@ -125,9 +124,14 @@ class AsyncConnection(ProxyComparable, StartableContext, AsyncConnectable):
         """Return the pooled DBAPI-level connection in use by this
         :class:`_asyncio.AsyncConnection`.
 
-        This is typically the SQLAlchemy connection-pool proxied connection
-        which then has an attribute .connection that refers to the actual
-        DBAPI-level connection.
+        This is a SQLAlchemy connection-pool proxied connection
+        which then has the attribute
+        :attr:`_pool._ConnectionFairy.driver_connection` that refers to the
+        actual driver connection. Its
+        :attr:`_pool._ConnectionFairy.dbapi_connection` refers instead
+        to an :class:`_engine.AdaptedConnection` instance that
+        adapts the driver connection to the DBAPI protocol.
+
         """
         conn = self._sync_connection()
 
diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index db63dfec8..38b0f67cb 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -52,6 +52,9 @@ class _ConnDialect(object):
             "passed to the connection pool."
         )
 
+    def get_driver_connection(self, connection):
+        return connection
+
 
 class _AsyncConnDialect(_ConnDialect):
     is_async = True
@@ -374,15 +377,63 @@ class _ConnectionRecord(object):
 
     starttime = None
 
-    connection = None
+    dbapi_connection = None
     """A reference to the actual DBAPI connection being tracked.
 
     May be ``None`` if this :class:`._ConnectionRecord` has been marked
     as invalidated; a new DBAPI connection may replace it if the owning
     pool calls upon this :class:`._ConnectionRecord` to reconnect.
 
+    For adapted drivers, like the Asyncio implementations, this is a
+    :class:`.AdaptedConnection` that adapts the driver connection
+    to the DBAPI protocol.
+    Use :attr:`._ConnectionRecord.driver_connection` to obtain the
+    connection objected returned by the driver.
+
+    .. versionadded:: 1.4.24
+
     """
 
+    @property
+    def driver_connection(self):
+        """The connection object as returned by the driver after a connect.
+
+        For normal sync drivers that support the DBAPI protocol, this object
+        is the same as the one referenced by
+        :attr:`._ConnectionRecord.dbapi_connection`.
+
+        For adapted drivers, like the Asyncio ones, this is the actual object
+        that was returned by the driver ``connect`` call.
+
+        As :attr:`._ConnectionRecord.dbapi_connection` it may be ``None``
+        if this :class:`._ConnectionRecord` has been marked as invalidated.
+
+        .. versionadded:: 1.4.24
+
+        """
+
+        if self.dbapi_connection is None:
+            return None
+        else:
+            return self.__pool._dialect.get_driver_connection(
+                self.dbapi_connection
+            )
+
+    @property
+    def connection(self):
+        """An alias to :attr:`._ConnectionRecord.dbapi_connection`.
+
+        This alias is deprecated, please use the new name.
+
+        .. deprecated:: 1.4.24
+
+        """
+        return self.dbapi_connection
+
+    @connection.setter
+    def connection(self, value):
+        self.dbapi_connection = value
+
     _soft_invalidate_time = 0
 
     @util.memoized_property
@@ -461,7 +512,7 @@ class _ConnectionRecord(object):
             util.warn("Double checkin attempted on %s" % self)
             return
         self.fairy_ref = None
-        connection = self.connection
+        connection = self.dbapi_connection
         pool = self.__pool
         while self.finalize_callback:
             finalizer = self.finalize_callback.pop()
@@ -480,11 +531,12 @@ class _ConnectionRecord(object):
         return self.starttime
 
     def close(self):
-        if self.connection is not None:
+        if self.dbapi_connection is not None:
             self.__close()
 
     def invalidate(self, e=None, soft=False):
-        """Invalidate the DBAPI connection held by this :class:`._ConnectionRecord`.
+        """Invalidate the DBAPI connection held by this
+        :class:`._ConnectionRecord`.
 
         This method is called for all connection invalidations, including
         when the :meth:`._ConnectionFairy.invalidate` or
@@ -492,10 +544,11 @@ class _ConnectionRecord(object):
         as well as when any
         so-called "automatic invalidation" condition occurs.
 
-        :param e: an exception object indicating a reason for the invalidation.
+        :param e: an exception object indicating a reason for the
+          invalidation.
 
         :param soft: if True, the connection isn't closed; instead, this
-         connection will be recycled on next checkout.
+          connection will be recycled on next checkout.
 
          .. versionadded:: 1.0.3
 
@@ -505,17 +558,19 @@ class _ConnectionRecord(object):
 
         """
         # already invalidated
-        if self.connection is None:
+        if self.dbapi_connection is None:
             return
         if soft:
-            self.__pool.dispatch.soft_invalidate(self.connection, self, e)
+            self.__pool.dispatch.soft_invalidate(
+                self.dbapi_connection, self, e
+            )
         else:
-            self.__pool.dispatch.invalidate(self.connection, self, e)
+            self.__pool.dispatch.invalidate(self.dbapi_connection, self, e)
         if e is not None:
             self.__pool.logger.info(
                 "%sInvalidate connection %r (reason: %s:%s)",
                 "Soft " if soft else "",
-                self.connection,
+                self.dbapi_connection,
                 e.__class__.__name__,
                 e,
             )
@@ -523,14 +578,14 @@ class _ConnectionRecord(object):
             self.__pool.logger.info(
                 "%sInvalidate connection %r",
                 "Soft " if soft else "",
-                self.connection,
+                self.dbapi_connection,
             )
 
         if soft:
             self._soft_invalidate_time = time.time()
         else:
             self.__close()
-            self.connection = None
+            self.dbapi_connection = None
 
     def get_connection(self):
         recycle = False
@@ -547,7 +602,7 @@ class _ConnectionRecord(object):
         # within 16 milliseconds accuracy, so unit tests for connection
         # invalidation need a sleep of at least this long between initial start
         # time and invalidation for the logic below to work reliably.
-        if self.connection is None:
+        if self.dbapi_connection is None:
             self.info.clear()
             self.__connect()
         elif (
@@ -555,21 +610,22 @@ class _ConnectionRecord(object):
             and time.time() - self.starttime > self.__pool._recycle
         ):
             self.__pool.logger.info(
-                "Connection %r exceeded timeout; recycling", self.connection
+                "Connection %r exceeded timeout; recycling",
+                self.dbapi_connection,
             )
             recycle = True
         elif self.__pool._invalidate_time > self.starttime:
             self.__pool.logger.info(
                 "Connection %r invalidated due to pool invalidation; "
                 + "recycling",
-                self.connection,
+                self.dbapi_connection,
             )
             recycle = True
         elif self._soft_invalidate_time > self.starttime:
             self.__pool.logger.info(
                 "Connection %r invalidated due to local soft invalidation; "
                 + "recycling",
-                self.connection,
+                self.dbapi_connection,
             )
             recycle = True
 
@@ -578,11 +634,11 @@ class _ConnectionRecord(object):
             self.info.clear()
 
             self.__connect()
-        return self.connection
+        return self.dbapi_connection
 
     def _is_hard_or_soft_invalidated(self):
         return (
-            self.connection is None
+            self.dbapi_connection is None
             or self.__pool._invalidate_time > self.starttime
             or (self._soft_invalidate_time > self.starttime)
         )
@@ -590,21 +646,20 @@ class _ConnectionRecord(object):
     def __close(self):
         self.finalize_callback.clear()
         if self.__pool.dispatch.close:
-            self.__pool.dispatch.close(self.connection, self)
-        self.__pool._close_connection(self.connection)
-        self.connection = None
+            self.__pool.dispatch.close(self.dbapi_connection, self)
+        self.__pool._close_connection(self.dbapi_connection)
+        self.dbapi_connection = None
 
     def __connect(self):
         pool = self.__pool
 
         # ensure any existing connection is removed, so that if
         # creator fails, this attribute stays None
-        self.connection = None
+        self.dbapi_connection = None
         try:
             self.starttime = time.time()
-            connection = pool._invoke_creator(self)
+            self.dbapi_connection = connection = pool._invoke_creator(self)
             pool.logger.debug("Created new connection %r", connection)
-            self.connection = connection
             self.fresh = True
         except Exception as e:
             with util.safe_reraise():
@@ -615,17 +670,17 @@ class _ConnectionRecord(object):
             if pool.dispatch.first_connect:
                 pool.dispatch.first_connect.for_modify(
                     pool.dispatch
-                ).exec_once_unless_exception(self.connection, self)
+                ).exec_once_unless_exception(self.dbapi_connection, self)
 
             # init of the dialect now takes place within the connect
             # event, so ensure a mutex is used on the first run
             pool.dispatch.connect.for_modify(
                 pool.dispatch
-            )._exec_w_sync_on_first_run(self.connection, self)
+            )._exec_w_sync_on_first_run(self.dbapi_connection, self)
 
 
 def _finalize_fairy(
-    connection,
+    dbapi_connection,
     connection_record,
     pool,
     ref,  # this is None when called directly, not by the gc
@@ -650,8 +705,8 @@ def _finalize_fairy(
     if ref is not None:
         if connection_record.fairy_ref is not ref:
             return
-        assert connection is None
-        connection = connection_record.connection
+        assert dbapi_connection is None
+        dbapi_connection = connection_record.dbapi_connection
 
     # null pool is not _is_asyncio but can be used also with async dialects
     dont_restore_gced = pool._dialect.is_async
@@ -663,11 +718,11 @@ def _finalize_fairy(
         detach = not connection_record
         can_manipulate_connection = True
 
-    if connection is not None:
+    if dbapi_connection is not None:
         if connection_record and echo:
             pool.logger.debug(
                 "Connection %r being returned to pool%s",
-                connection,
+                dbapi_connection,
                 ", transaction state was already reset by caller"
                 if not reset
                 else "",
@@ -675,9 +730,11 @@ def _finalize_fairy(
 
         try:
             fairy = fairy or _ConnectionFairy(
-                connection, connection_record, echo
+                dbapi_connection,
+                connection_record,
+                echo,
             )
-            assert fairy.connection is connection
+            assert fairy.dbapi_connection is dbapi_connection
             if reset and can_manipulate_connection:
                 fairy._reset(pool)
 
@@ -688,9 +745,9 @@ def _finalize_fairy(
 
                 if can_manipulate_connection:
                     if pool.dispatch.close_detached:
-                        pool.dispatch.close_detached(connection)
+                        pool.dispatch.close_detached(dbapi_connection)
 
-                    pool._close_connection(connection)
+                    pool._close_connection(dbapi_connection)
                 else:
                     message = (
                         "The garbage collector is trying to clean up "
@@ -700,7 +757,7 @@ def _finalize_fairy(
                         "connections when they are no longer used, calling "
                         "``close()`` or using a context manager to "
                         "manage their lifetime."
-                    ) % connection
+                    ) % dbapi_connection
                     pool.logger.error(message)
                     util.warn(message)
 
@@ -746,12 +803,24 @@ class _ConnectionFairy(object):
     """
 
     def __init__(self, dbapi_connection, connection_record, echo):
-        self.connection = dbapi_connection
+        self.dbapi_connection = dbapi_connection
         self._connection_record = connection_record
         self._echo = echo
 
-    connection = None
-    """A reference to the actual DBAPI connection being tracked."""
+    dbapi_connection = None
+    """A reference to the actual DBAPI connection being tracked.
+
+    .. versionadded:: 1.4.24
+
+    .. seealso::
+
+        :attr:`._ConnectionFairy.driver_connection`
+
+        :attr:`._ConnectionRecord.dbapi_connection`
+
+        :ref:`faq_dbapi_connection`
+
+    """
 
     _connection_record = None
     """A reference to the :class:`._ConnectionRecord` object associated
@@ -761,6 +830,38 @@ class _ConnectionFairy(object):
 
     """
 
+    @property
+    def driver_connection(self):
+        """The connection object as returned by the driver after a connect.
+
+        .. versionadded:: 1.4.24
+
+        .. seealso::
+
+            :attr:`._ConnectionFairy.dbapi_connection`
+
+            :attr:`._ConnectionRecord.driver_connection`
+
+            :ref:`faq_dbapi_connection`
+
+        """
+        return self._connection_record.driver_connection
+
+    @property
+    def connection(self):
+        """An alias to :attr:`._ConnectionFairy.dbapi_connection`.
+
+        This alias is deprecated, please use the new name.
+
+        .. deprecated:: 1.4.24
+
+        """
+        return self.dbapi_connection
+
+    @connection.setter
+    def connection(self, value):
+        self.dbapi_connection = value
+
     @classmethod
     def _checkout(cls, pool, threadconns=None, fairy=None):
         if not fairy:
@@ -772,7 +873,7 @@ class _ConnectionFairy(object):
             if threadconns is not None:
                 threadconns.current = weakref.ref(fairy)
 
-        if fairy.connection is None:
+        if fairy.dbapi_connection is None:
             raise exc.InvalidRequestError("This connection is closed")
         fairy._counter += 1
         if (
@@ -795,25 +896,25 @@ class _ConnectionFairy(object):
                         if fairy._echo:
                             pool.logger.debug(
                                 "Pool pre-ping on connection %s",
-                                fairy.connection,
+                                fairy.dbapi_connection,
                             )
-                        result = pool._dialect.do_ping(fairy.connection)
+                        result = pool._dialect.do_ping(fairy.dbapi_connection)
                         if not result:
                             if fairy._echo:
                                 pool.logger.debug(
                                     "Pool pre-ping on connection %s failed, "
                                     "will invalidate pool",
-                                    fairy.connection,
+                                    fairy.dbapi_connection,
                                 )
                             raise exc.InvalidatePoolError()
                     elif fairy._echo:
                         pool.logger.debug(
                             "Connection %s is fresh, skipping pre-ping",
-                            fairy.connection,
+                            fairy.dbapi_connection,
                         )
 
                 pool.dispatch.checkout(
-                    fairy.connection, fairy._connection_record, fairy
+                    fairy.dbapi_connection, fairy._connection_record, fairy
                 )
                 return fairy
             except exc.DisconnectionError as e:
@@ -830,12 +931,12 @@ class _ConnectionFairy(object):
                     pool.logger.info(
                         "Disconnection detected on checkout, "
                         "invalidating individual connection %s (reason: %r)",
-                        fairy.connection,
+                        fairy.dbapi_connection,
                         e,
                     )
                     fairy._connection_record.invalidate(e)
                 try:
-                    fairy.connection = (
+                    fairy.dbapi_connection = (
                         fairy._connection_record.get_connection()
                     )
                 except Exception as err:
@@ -863,7 +964,7 @@ class _ConnectionFairy(object):
 
     def _checkin(self, reset=True):
         _finalize_fairy(
-            self.connection,
+            self.dbapi_connection,
             self._connection_record,
             self._pool,
             None,
@@ -871,7 +972,7 @@ class _ConnectionFairy(object):
             reset=reset,
             fairy=self,
         )
-        self.connection = None
+        self.dbapi_connection = None
         self._connection_record = None
 
     _close = _checkin
@@ -882,14 +983,14 @@ class _ConnectionFairy(object):
         if pool._reset_on_return is reset_rollback:
             if self._echo:
                 pool.logger.debug(
-                    "Connection %s rollback-on-return", self.connection
+                    "Connection %s rollback-on-return", self.dbapi_connection
                 )
             pool._dialect.do_rollback(self)
         elif pool._reset_on_return is reset_commit:
             if self._echo:
                 pool.logger.debug(
                     "Connection %s commit-on-return",
-                    self.connection,
+                    self.dbapi_connection,
                 )
             pool._dialect.do_commit(self)
 
@@ -902,7 +1003,7 @@ class _ConnectionFairy(object):
         """Return True if this :class:`._ConnectionFairy` still refers
         to an active DBAPI connection."""
 
-        return self.connection is not None
+        return self.dbapi_connection is not None
 
     @util.memoized_property
     def info(self):
@@ -963,13 +1064,13 @@ class _ConnectionFairy(object):
 
         """
 
-        if self.connection is None:
+        if self.dbapi_connection is None:
             util.warn("Can't invalidate an already-closed connection.")
             return
         if self._connection_record:
             self._connection_record.invalidate(e=e, soft=soft)
         if not soft:
-            self.connection = None
+            self.dbapi_connection = None
             self._checkin()
 
     def cursor(self, *args, **kwargs):
@@ -979,10 +1080,10 @@ class _ConnectionFairy(object):
         method.
 
         """
-        return self.connection.cursor(*args, **kwargs)
+        return self.dbapi_connection.cursor(*args, **kwargs)
 
     def __getattr__(self, key):
-        return getattr(self.connection, key)
+        return getattr(self.dbapi_connection, key)
 
     def detach(self):
         """Separate this connection from its Pool.
@@ -1000,14 +1101,14 @@ class _ConnectionFairy(object):
         if self._connection_record is not None:
             rec = self._connection_record
             rec.fairy_ref = None
-            rec.connection = None
+            rec.dbapi_connection = None
             # TODO: should this be _return_conn?
             self._pool._do_return_conn(self._connection_record)
             self.info = self.info.copy()
             self._connection_record = None
 
             if self._pool.dispatch.detach:
-                self._pool.dispatch.detach(self.connection, rec)
+                self._pool.dispatch.detach(self.dbapi_connection, rec)
 
     def close(self):
         self._counter -= 1
diff --git a/lib/sqlalchemy/pool/events.py b/lib/sqlalchemy/pool/events.py
index 18ef28fa5..7c2cae7c5 100644
--- a/lib/sqlalchemy/pool/events.py
+++ b/lib/sqlalchemy/pool/events.py
@@ -71,6 +71,7 @@ class PoolEvents(event.Events):
         to produce a new DBAPI connection.
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -95,6 +96,7 @@ class PoolEvents(event.Events):
         encoding settings, collation settings, and many others.
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -105,6 +107,7 @@ class PoolEvents(event.Events):
         """Called when a connection is retrieved from the Pool.
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -132,6 +135,7 @@ class PoolEvents(event.Events):
         for detached connections.  (They do not return to the pool.)
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -153,6 +157,7 @@ class PoolEvents(event.Events):
         cases where the connection is discarded immediately after reset.
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -176,6 +181,7 @@ class PoolEvents(event.Events):
         connection occurs.
 
         :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
 
         :param connection_record: the :class:`._ConnectionRecord` managing the
          DBAPI connection.
@@ -205,6 +211,15 @@ class PoolEvents(event.Events):
 
         .. versionadded:: 1.0.3
 
+        :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
+
+        :param connection_record: the :class:`._ConnectionRecord` managing the
+         DBAPI connection.
+
+        :param exception: the exception object corresponding to the reason
+         for this invalidation, if any.  May be ``None``.
+
         """
 
     def close(self, dbapi_connection, connection_record):
@@ -222,6 +237,12 @@ class PoolEvents(event.Events):
 
         .. versionadded:: 1.1
 
+        :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
+
+        :param connection_record: the :class:`._ConnectionRecord` managing the
+         DBAPI connection.
+
         """
 
     def detach(self, dbapi_connection, connection_record):
@@ -232,6 +253,12 @@ class PoolEvents(event.Events):
 
         .. versionadded:: 1.1
 
+        :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
+
+        :param connection_record: the :class:`._ConnectionRecord` managing the
+         DBAPI connection.
+
         """
 
     def close_detached(self, dbapi_connection):
@@ -245,4 +272,7 @@ class PoolEvents(event.Events):
 
         .. versionadded:: 1.1
 
+        :param dbapi_connection: a DBAPI connection.
+         The :attr:`._ConnectionRecord.dbapi_connection` attribute.
+
         """
diff --git a/lib/sqlalchemy/pool/impl.py b/lib/sqlalchemy/pool/impl.py
index 8a3412385..3ef33d02d 100644
--- a/lib/sqlalchemy/pool/impl.py
+++ b/lib/sqlalchemy/pool/impl.py
@@ -410,7 +410,7 @@ class StaticPool(Pool):
     def dispose(self):
         if (
             "connection" in self.__dict__
-            and self.connection.connection is not None
+            and self.connection.dbapi_connection is not None
         ):
             self.connection.close()
             del self.__dict__["connection"]
@@ -432,7 +432,7 @@ class StaticPool(Pool):
         # used by the test suite to make a new engine / pool without
         # losing the state of an existing SQLite :memory: connection
         self._invoke_creator = (
-            lambda crec: other_static_pool.connection.connection
+            lambda crec: other_static_pool.connection.dbapi_connection
         )
 
     def _create_connection(self):
diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 7d78dcc1a..a54f70c5e 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -65,7 +65,7 @@ class ConnectionKiller(object):
 
         for rec in list(self.proxy_refs):
             if rec is not None and rec.is_valid:
-                self.dbapi_connections.discard(rec.connection)
+                self.dbapi_connections.discard(rec.dbapi_connection)
                 self._safe(rec._checkin)
 
         # for fairy refs that were GCed and could not close the connection,
diff --git a/lib/sqlalchemy/testing/plugin/pytestplugin.py b/lib/sqlalchemy/testing/plugin/pytestplugin.py
index 4e82f10c1..d28048f70 100644
--- a/lib/sqlalchemy/testing/plugin/pytestplugin.py
+++ b/lib/sqlalchemy/testing/plugin/pytestplugin.py
@@ -358,6 +358,7 @@ _current_class = None
 
 def pytest_runtest_setup(item):
     from sqlalchemy.testing import asyncio
+    from sqlalchemy.util import string_types
 
     if not isinstance(item, pytest.Function):
         return
@@ -378,13 +379,38 @@ def pytest_runtest_setup(item):
         _current_class = item.parent.parent
 
         def finalize():
-            global _current_class
+            global _current_class, _current_report
             _current_class = None
 
-            asyncio._maybe_async_provisioning(
-                plugin_base.stop_test_class_outside_fixtures,
-                item.parent.parent.cls,
-            )
+            try:
+                asyncio._maybe_async_provisioning(
+                    plugin_base.stop_test_class_outside_fixtures,
+                    item.parent.parent.cls,
+                )
+            except Exception as e:
+                # in case of an exception during teardown attach the original
+                # error to the exception message, otherwise it will get lost
+                if _current_report.failed:
+                    if not e.args:
+                        e.args = (
+                            "__Original test failure__:\n"
+                            + _current_report.longreprtext,
+                        )
+                    elif e.args[-1] and isinstance(e.args[-1], string_types):
+                        args = list(e.args)
+                        args[-1] += (
+                            "\n__Original test failure__:\n"
+                            + _current_report.longreprtext
+                        )
+                        e.args = tuple(args)
+                    else:
+                        e.args += (
+                            "__Original test failure__",
+                            _current_report.longreprtext,
+                        )
+                raise
+            finally:
+                _current_report = None
 
         item.parent.parent.addfinalizer(finalize)
 
@@ -404,6 +430,15 @@ def pytest_runtest_call(item):
     )
 
 
+_current_report = None
+
+
+def pytest_runtest_logreport(report):
+    global _current_report
+    if report.when == "call":
+        _current_report = report
+
+
 def pytest_runtest_teardown(item, nextitem):
     # runs inside of pytest function fixture scope
     # after test function runs