- The engine-level error handling and wrapping routines will now

take effect in all engine connection use cases, including
when user-custom connect routines are used via the
:paramref:`.create_engine.creator` parameter, as well as when
the :class:`.Connection` encounters a connection error on
revalidation.
fixes #3266

8 files changed, 243 insertions, 19 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index b71ecc15d..b8b513821 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -23,6 +23,21 @@
     on compatibility concerns, see :doc:`/changelog/migration_10`.
 
     .. change::
+        :tags: bug, engine
+        :tickets: 3266
+
+        The engine-level error handling and wrapping routines will now
+        take effect in all engine connection use cases, including
+        when user-custom connect routines are used via the
+        :paramref:`.create_engine.creator` parameter, as well as when
+        the :class:`.Connection` encounters a connection error on
+        revalidation.
+
+        .. seealso::
+
+            :ref:`change_3266`
+
+    .. change::
         :tags: feature, oracle
 
         New Oracle DDL features for tables, indexes: COMPRESS, BITMAP.
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index 27a4fae4c..15e066a75 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -872,6 +872,29 @@ labeled uniquely.
 
 :ticket:`3170`
 
+.. _change_3266:
+
+DBAPI exception wrapping and handle_error() event improvements
+--------------------------------------------------------------
+
+SQLAlchemy's wrapping of DBAPI exceptions was not taking place in the
+case where a :class:`.Connection` object was invalidated, and then tried
+to reconnect and encountered an error; this has been resolved.
+
+Additionally, the recently added :meth:`.ConnectionEvents.handle_error`
+event is now invoked for errors that occur upon initial connect, upon
+reconnect, and when :func:`.create_engine` is used given a custom connection
+function via :paramref:`.create_engine.creator`.
+
+The :class:`.ExceptionContext` object has a new datamember
+:attr:`.ExceptionContext.engine` that will always refer to the :class:`.Engine`
+in use, in those cases when the :class:`.Connection` object is not available
+(e.g. on initial connect).
+
+
+:ticket:`3266`
+
+
 .. _behavioral_changes_orm_10:
 
 Behavioral Changes - ORM
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index dd82be1d1..901ab07eb 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -276,7 +276,7 @@ class Connection(Connectable):
                 raise exc.InvalidRequestError(
                     "Can't reconnect until invalid "
                     "transaction is rolled back")
-            self.__connection = self.engine.raw_connection()
+            self.__connection = self.engine.raw_connection(self)
             self.__invalid = False
             return self.__connection
         raise exc.ResourceClosedError("This Connection is closed")
@@ -1194,7 +1194,8 @@ class Connection(Connectable):
 
                 # new handle_error event
                 ctx = ExceptionContextImpl(
-                    e, sqlalchemy_exception, self, cursor, statement,
+                    e, sqlalchemy_exception, self.engine,
+                    self, cursor, statement,
                     parameters, context, self._is_disconnect)
 
                 for fn in self.dispatch.handle_error:
@@ -1242,6 +1243,58 @@ class Connection(Connectable):
             if self.should_close_with_result:
                 self.close()
 
+    @classmethod
+    def _handle_dbapi_exception_noconnection(
+            cls, e, dialect, engine, connection):
+        exc_info = sys.exc_info()
+
+        is_disconnect = dialect.is_disconnect(e, None, None)
+
+        should_wrap = isinstance(e, dialect.dbapi.Error)
+
+        if should_wrap:
+            sqlalchemy_exception = exc.DBAPIError.instance(
+                None,
+                None,
+                e,
+                dialect.dbapi.Error,
+                connection_invalidated=is_disconnect)
+        else:
+            sqlalchemy_exception = None
+
+        newraise = None
+
+        if engine._has_events:
+            ctx = ExceptionContextImpl(
+                e, sqlalchemy_exception, engine, connection, None, None,
+                None, None, is_disconnect)
+            for fn in engine.dispatch.handle_error:
+                try:
+                    # handler returns an exception;
+                    # call next handler in a chain
+                    per_fn = fn(ctx)
+                    if per_fn is not None:
+                        ctx.chained_exception = newraise = per_fn
+                except Exception as _raised:
+                    # handler raises an exception - stop processing
+                    newraise = _raised
+                    break
+
+            if sqlalchemy_exception and \
+                    is_disconnect != ctx.is_disconnect:
+                sqlalchemy_exception.connection_invalidated = \
+                    is_disconnect = ctx.is_disconnect
+
+        if newraise:
+            util.raise_from_cause(newraise, exc_info)
+        elif should_wrap:
+            util.raise_from_cause(
+                sqlalchemy_exception,
+                exc_info
+            )
+        else:
+            util.reraise(*exc_info)
+
     def default_schema_name(self):
         return self.engine.dialect.get_default_schema_name(self)
 
@@ -1320,8 +1373,9 @@ class ExceptionContextImpl(ExceptionContext):
     """Implement the :class:`.ExceptionContext` interface."""
 
     def __init__(self, exception, sqlalchemy_exception,
-                 connection, cursor, statement, parameters,
+                 engine, connection, cursor, statement, parameters,
                  context, is_disconnect):
+        self.engine = engine
         self.connection = connection
         self.sqlalchemy_exception = sqlalchemy_exception
         self.original_exception = exception
@@ -1898,7 +1952,15 @@ class Engine(Connectable, log.Identified):
         """
         return self.run_callable(self.dialect.has_table, table_name, schema)
 
-    def raw_connection(self):
+    def _wrap_pool_connect(self, fn, connection=None):
+        dialect = self.dialect
+        try:
+            return fn()
+        except dialect.dbapi.Error as e:
+            Connection._handle_dbapi_exception_noconnection(
+                e, dialect, self, connection)
+
+    def raw_connection(self, _connection=None):
         """Return a "raw" DBAPI connection from the connection pool.
 
         The returned object is a proxied version of the DBAPI
@@ -1914,8 +1976,8 @@ class Engine(Connectable, log.Identified):
         :meth:`.Engine.connect` method.
 
         """
-
-        return self.pool.unique_connection()
+        return self._wrap_pool_connect(
+            self.pool.unique_connection, _connection)
 
 
 class OptionEngine(Engine):
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 0ad2efae0..5f66e54b5 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -917,7 +917,23 @@ class ExceptionContext(object):
     connection = None
     """The :class:`.Connection` in use during the exception.
 
-    This member is always present.
+    This member is present, except in the case of a failure when
+    first connecting.
+
+    .. seealso::
+
+        :attr:`.ExceptionContext.engine`
+
+
+    """
+
+    engine = None
+    """The :class:`.Engine` in use during the exception.
+
+    This member should always be present, even in the case of a failure
+    when first connecting.
+
+    .. versionadded:: 1.0.0
 
     """
 
diff --git a/lib/sqlalchemy/engine/strategies.py b/lib/sqlalchemy/engine/strategies.py
index 398ef8df6..fd665ad03 100644
--- a/lib/sqlalchemy/engine/strategies.py
+++ b/lib/sqlalchemy/engine/strategies.py
@@ -86,16 +86,7 @@ class DefaultEngineStrategy(EngineStrategy):
         pool = pop_kwarg('pool', None)
         if pool is None:
             def connect():
-                try:
-                    return dialect.connect(*cargs, **cparams)
-                except dialect.dbapi.Error as e:
-                    invalidated = dialect.is_disconnect(e, None, None)
-                    util.raise_from_cause(
-                        exc.DBAPIError.instance(
-                            None, None, e, dialect.dbapi.Error,
-                            connection_invalidated=invalidated
-                        )
-                    )
+                return dialect.connect(*cargs, **cparams)
 
             creator = pop_kwarg('creator', connect)
 
diff --git a/lib/sqlalchemy/engine/threadlocal.py b/lib/sqlalchemy/engine/threadlocal.py
index 637523a0e..71caac626 100644
--- a/lib/sqlalchemy/engine/threadlocal.py
+++ b/lib/sqlalchemy/engine/threadlocal.py
@@ -59,7 +59,7 @@ class TLEngine(base.Engine):
             # guards against pool-level reapers, if desired.
             # or not connection.connection.is_valid:
             connection = self._tl_connection_cls(
-                self, self.pool.connect(), **kw)
+                self, self._wrap_pool_connect(self.pool.connect), **kw)
             self._connections.conn = weakref.ref(connection)
 
         return connection._increment_connect()
diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py
index c144902cd..8600c20f5 100644
--- a/lib/sqlalchemy/events.py
+++ b/lib/sqlalchemy/events.py
@@ -739,6 +739,12 @@ class ConnectionEvents(event.Events):
         .. versionadded:: 0.9.7 Added the
             :meth:`.ConnectionEvents.handle_error` hook.
 
+        .. versionchanged:: 1.0.0 The :meth:`.handle_error` event is now
+           invoked when an :class:`.Engine` fails during the initial
+           call to :meth:`.Engine.connect`, as well as when a
+           :class:`.Connection` object encounters an error during a
+           reconnect operation.
+
         .. versionchanged:: 1.0.0 The :meth:`.handle_error` event is
            not fired off when a dialect makes use of the
            ``skip_user_error_events`` execution option.   This is used
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index d8f202f99..72a089aca 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -6,6 +6,7 @@ import sqlalchemy as tsa
 from sqlalchemy.testing import fixtures
 from sqlalchemy import testing
 from sqlalchemy.testing.mock import Mock, MagicMock
+from sqlalchemy import event
 
 dialect = None
 
@@ -240,7 +241,6 @@ class CreateEngineTest(fixtures.TestBase):
     def test_wraps_connect_in_dbapi(self):
         e = create_engine('sqlite://')
         sqlite3 = e.dialect.dbapi
-
         dbapi = MockDBAPI()
         dbapi.Error = sqlite3.Error,
         dbapi.ProgrammingError = sqlite3.ProgrammingError
@@ -253,6 +253,117 @@ class CreateEngineTest(fixtures.TestBase):
             assert not de.connection_invalidated
 
     @testing.requires.sqlite
+    def test_handle_error_event_connect(self):
+        e = create_engine('sqlite://')
+        dbapi = MockDBAPI()
+        sqlite3 = e.dialect.dbapi
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+        dbapi.connect = Mock(
+            side_effect=sqlite3.ProgrammingError("random error"))
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is None
+            raise MySpecialException("failed operation")
+
+        assert_raises(
+            MySpecialException,
+            eng.connect
+        )
+
+    @testing.requires.sqlite
+    def test_handle_error_event_reconnect(self):
+        e = create_engine('sqlite://')
+        dbapi = MockDBAPI()
+        sqlite3 = e.dialect.dbapi
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi, _initialize=False)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is conn
+            raise MySpecialException("failed operation")
+
+        conn = eng.connect()
+        conn.invalidate()
+
+        dbapi.connect = Mock(
+            side_effect=sqlite3.ProgrammingError("random error"))
+
+        assert_raises(
+            MySpecialException,
+            conn._revalidate_connection
+        )
+
+    @testing.requires.sqlite
+    def test_handle_error_custom_connect(self):
+        e = create_engine('sqlite://')
+
+        dbapi = MockDBAPI()
+        sqlite3 = e.dialect.dbapi
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+
+        class MySpecialException(Exception):
+            pass
+
+        def custom_connect():
+            raise sqlite3.ProgrammingError("random error")
+
+        eng = create_engine('sqlite://', module=dbapi, creator=custom_connect)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.engine is eng
+            assert ctx.connection is None
+            raise MySpecialException("failed operation")
+
+        assert_raises(
+            MySpecialException,
+            eng.connect
+        )
+
+    @testing.requires.sqlite
+    def test_handle_error_event_connect_invalidate_flag(self):
+        e = create_engine('sqlite://')
+        dbapi = MockDBAPI()
+        sqlite3 = e.dialect.dbapi
+        dbapi.Error = sqlite3.Error,
+        dbapi.ProgrammingError = sqlite3.ProgrammingError
+        dbapi.connect = Mock(
+            side_effect=sqlite3.ProgrammingError(
+                "Cannot operate on a closed database."))
+
+        class MySpecialException(Exception):
+            pass
+
+        eng = create_engine('sqlite://', module=dbapi)
+
+        @event.listens_for(eng, "handle_error")
+        def handle_error(ctx):
+            assert ctx.is_disconnect
+            ctx.is_disconnect = False
+
+        try:
+            eng.connect()
+            assert False
+        except tsa.exc.DBAPIError as de:
+            assert not de.connection_invalidated
+
+    @testing.requires.sqlite
     def test_dont_touch_non_dbapi_exception_on_connect(self):
         e = create_engine('sqlite://')
         sqlite3 = e.dialect.dbapi