- The mechanics of the :meth:`.ConnectionEvents.dbapi_error` handler

have been enhanced such that the function handler is now capable
of raising or returning a new exception object, which will replace
the exception normally being thrown by SQLAlchemy.
fixes #3076

5 files changed, 239 insertions, 25 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index f3389ffda..074dd6e31 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -15,6 +15,16 @@
     :released:
 
     .. change::
+        :tags: feature, engine
+        :tickets: 3076
+        :versions: 1.0.0
+
+        The mechanics of the :meth:`.ConnectionEvents.dbapi_error` handler
+        have been enhanced such that the function handler is now capable
+        of raising or returning a new exception object, which will replace
+        the exception normally being thrown by SQLAlchemy.
+
+    .. change::
         :tags: bug, orm
         :tickets: 3108
         :versions: 1.0.0
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index 43830197e..b9c45642c 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -131,6 +131,34 @@ wishes to support the new feature should now call upon the ``._limit_clause``
 and ``._offset_clause`` attributes to receive the full SQL expression, rather
 than the integer value.
 
+.. _feature_3076:
+
+DBAPI Exceptions may be re-stated using events
+----------------------------------------------
+
+.. note::
+
+	this feature is also back-ported to SQLAlchemy 0.9.7.
+
+The :meth:`.ConnectionEvents.dbapi_error` handler may now be used to re-state
+the exception raised as an alternate, user-defined exception::
+
+    @event.listens_for(Engine, "dbapi_error")
+    def handle_exception(conn, cursor, statement, parameters, context, exception):
+        if isinstance(exception, psycopg2.OperationalError) and \
+            "failed" in str(exception):
+            raise MySpecialException("failed operation")
+
+The handler supports both raising an exception immediately, as well
+as being able to return the new exception such that the chain of event handling
+will continue, the next event handler receiving the new exception as
+its argument.
+
+:ticket:`3076`
+
+.. seealso::
+
+	:meth:`.ConnectionEvents.dbapi_error`
 
 Behavioral Improvements
 =======================
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 249c494fe..c885bcf69 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -1090,14 +1090,28 @@ class Connection(Connectable):
             should_wrap = isinstance(e, self.dialect.dbapi.Error) or \
                 (statement is not None and context is None)
 
+            newraise = None
             if should_wrap and context:
                 if self._has_events or self.engine._has_events:
-                    self.dispatch.dbapi_error(self,
-                                                    cursor,
-                                                    statement,
-                                                    parameters,
-                                                    context,
-                                                    e)
+                    for fn in self.dispatch.dbapi_error:
+                        try:
+                            # handler returns an exception;
+                            # call next handler in a chain
+                            per_fn = fn(self,
+                                                cursor,
+                                                statement,
+                                                parameters,
+                                                context,
+                                                newraise
+                                                    if newraise
+                                                    is not None else e)
+                            if per_fn is not None:
+                                newraise = per_fn
+                        except Exception as _raised:
+                            # handler raises an exception - stop processing
+                            newraise = _raised
+
+
                 context.handle_dbapi_exception(e)
 
             if not self._is_disconnect:
@@ -1105,7 +1119,9 @@ class Connection(Connectable):
                     self._safe_close_cursor(cursor)
                 self._autorollback()
 
-            if should_wrap:
+            if newraise:
+                util.raise_from_cause(newraise, exc_info)
+            elif should_wrap:
                 util.raise_from_cause(
                                     exc.DBAPIError.instance(
                                         statement,
diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py
index 908ce378f..d3fe80b76 100644
--- a/lib/sqlalchemy/events.py
+++ b/lib/sqlalchemy/events.py
@@ -492,12 +492,12 @@ class ConnectionEvents(event.Events):
                         parameters, context, executemany)
                     return statement, parameters
                 fn = wrap_before_cursor_execute
-
         elif retval and \
-            identifier not in ('before_execute', 'before_cursor_execute'):
+            identifier not in ('before_execute',
+                                    'before_cursor_execute', 'dbapi_error'):
             raise exc.ArgumentError(
-                    "Only the 'before_execute' and "
-                    "'before_cursor_execute' engine "
+                    "Only the 'before_execute', "
+                    "'before_cursor_execute' and 'dbapi_error' engine "
                     "event listeners accept the 'retval=True' "
                     "argument.")
         event_key.with_wrapper(fn).base_listen()
@@ -616,20 +616,69 @@ class ConnectionEvents(event.Events):
         existing transaction remains in effect as well as any state
         on the cursor.
 
-        The use case here is to inject low-level exception handling
-        into an :class:`.Engine`, typically for logging and
-        debugging purposes.   In general, user code should **not** modify
-        any state or throw any exceptions here as this will
-        interfere with SQLAlchemy's cleanup and error handling
-        routines.
-
-        Subsequent to this hook, SQLAlchemy may attempt any
-        number of operations on the connection/cursor, including
-        closing the cursor, rolling back of the transaction in the
-        case of connectionless execution, and disposing of the entire
-        connection pool if a "disconnect" was detected.   The
-        exception is then wrapped in a SQLAlchemy DBAPI exception
-        wrapper and re-thrown.
+        The use cases supported by this hook include:
+
+        * read-only, low-level exception handling for logging and
+          debugging purposes
+        * exception re-writing (0.9.7 and up only)
+
+        The hook is called while the cursor from the failed operation
+        (if any) is still open and accessible.   Special cleanup operations
+        can be called on this cursor; SQLAlchemy will attempt to close
+        this cursor subsequent to this hook being invoked.  If the connection
+        is in "autocommit" mode, the transaction also remains open within
+        the scope of this hook; the rollback of the per-statement transaction
+        also occurs after the hook is called.
+
+        When cleanup operations are complete, SQLAlchemy wraps the DBAPI-specific
+        exception in a SQLAlchemy-level wrapper mirroring the exception class,
+        and then propagates that new exception object.
+
+        The user-defined event handler has two options for replacing
+        the SQLAlchemy-constructed exception into one that is user
+        defined.   It can either raise this new exception directly, in
+        which case all further event listeners are bypassed and the
+        exception will be raised, after appropriate cleanup as taken
+        place::
+
+            # 0.9.7 and up only !!!
+            @event.listens_for(Engine, "dbapi_error")
+            def handle_exception(conn, cursor, statement, parameters, context, exception):
+                if isinstance(exception, psycopg2.OperationalError) and \
+                    "failed" in str(exception):
+                    raise MySpecialException("failed operation")
+
+        Alternatively, a "chained" style of event handling can be
+        used, by configuring the handler with the ``retval=True``
+        modifier and returning the new exception instance from the
+        function.  In this case, event handling will continue onto the
+        next handler, that handler receiving the new exception as its
+        argument::
+
+            # 0.9.7 and up only !!!
+            @event.listens_for(Engine, "dbapi_error", retval=True)
+            def handle_exception(conn, cursor, statement, parameters, context, exception):
+                if isinstance(exception, psycopg2.OperationalError) and \
+                    "failed" in str(exception):
+                    return MySpecialException("failed operation")
+                else:
+                    return None
+
+        Handlers that return ``None`` may remain within this chain; the
+        last non-``None`` return value is the one that continues to be
+        passed to the next handler.
+
+        When a custom exception is raised or returned, SQLAlchemy raises
+        this new exception as-is, it is not wrapped by any SQLAlchemy
+        object.  If the exception is not a subclass of
+        :class:`sqlalchemy.exc.StatementError`,
+        certain features may not be available; currently this includes
+        the ORM's feature of adding a detail hint about "autoflush" to
+        exceptions raised within the autoflush process.
+
+        .. versionadded:: 0.9.7 Support for translation of DBAPI exceptions
+           into user-defined exceptions within the
+           :meth:`.ConnectionEvents.dbapi_error` event hook.
 
         :param conn: :class:`.Connection` object
         :param cursor: DBAPI cursor object
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 78ae40460..7e7126fce 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1131,6 +1131,117 @@ class EngineEventsTest(fixtures.TestBase):
             assert canary[0][2] is e.orig
             assert canary[0][0] == "SELECT FOO FROM I_DONT_EXIST"
 
+    def test_exception_event_reraise(self):
+        engine = engines.testing_engine()
+
+        class MyException(Exception):
+            pass
+
+        @event.listens_for(engine, 'dbapi_error', retval=True)
+        def err(conn, cursor, stmt, parameters, context, exception):
+            if "ERROR ONE" in str(stmt):
+                return MyException("my exception")
+            elif "ERROR TWO" in str(stmt):
+                return exception
+            else:
+                return None
+
+        conn = engine.connect()
+        # case 1: custom exception
+        assert_raises_message(
+            MyException,
+            "my exception",
+            conn.execute, "SELECT 'ERROR ONE' FROM I_DONT_EXIST"
+        )
+        # case 2: return the DBAPI exception we're given;
+        # no wrapping should occur
+        assert_raises(
+            conn.dialect.dbapi.Error,
+            conn.execute, "SELECT 'ERROR TWO' FROM I_DONT_EXIST"
+        )
+        # case 3: normal wrapping
+        assert_raises(
+            tsa.exc.DBAPIError,
+            conn.execute, "SELECT 'ERROR THREE' FROM I_DONT_EXIST"
+        )
+
+    def test_exception_event_reraise_chaining(self):
+        engine = engines.testing_engine()
+
+        class MyException1(Exception):
+            pass
+
+        class MyException2(Exception):
+            pass
+
+        class MyException3(Exception):
+            pass
+
+        @event.listens_for(engine, 'dbapi_error', retval=True)
+        def err1(conn, cursor, stmt, parameters, context, exception):
+            if "ERROR ONE" in str(stmt) or "ERROR TWO" in str(stmt) \
+                    or "ERROR THREE" in str(stmt):
+                return MyException1("my exception")
+            elif "ERROR FOUR" in str(stmt):
+                raise MyException3("my exception short circuit")
+
+        @event.listens_for(engine, 'dbapi_error', retval=True)
+        def err2(conn, cursor, stmt, parameters, context, exception):
+            if ("ERROR ONE" in str(stmt) or "ERROR FOUR" in str(stmt)) \
+                and isinstance(exception, MyException1):
+                raise MyException2("my exception chained")
+            elif "ERROR TWO" in str(stmt):
+                return exception
+            else:
+                return None
+
+        conn = engine.connect()
+
+        with patch.object(engine.
+                dialect.execution_ctx_cls, "handle_dbapi_exception") as patched:
+            assert_raises_message(
+                MyException2,
+                "my exception chained",
+                conn.execute, "SELECT 'ERROR ONE' FROM I_DONT_EXIST"
+            )
+            eq_(patched.call_count, 1)
+
+        with patch.object(engine.
+                dialect.execution_ctx_cls, "handle_dbapi_exception") as patched:
+            assert_raises(
+                MyException1,
+                conn.execute, "SELECT 'ERROR TWO' FROM I_DONT_EXIST"
+            )
+            eq_(patched.call_count, 1)
+
+        with patch.object(engine.
+                dialect.execution_ctx_cls, "handle_dbapi_exception") as patched:
+            # test that non None from err1 isn't cancelled out
+            # by err2
+            assert_raises(
+                MyException1,
+                conn.execute, "SELECT 'ERROR THREE' FROM I_DONT_EXIST"
+            )
+            eq_(patched.call_count, 1)
+
+        with patch.object(engine.
+                dialect.execution_ctx_cls, "handle_dbapi_exception") as patched:
+            assert_raises(
+                tsa.exc.DBAPIError,
+                conn.execute, "SELECT 'ERROR FIVE' FROM I_DONT_EXIST"
+            )
+            eq_(patched.call_count, 1)
+
+        with patch.object(engine.
+                dialect.execution_ctx_cls, "handle_dbapi_exception") as patched:
+            assert_raises_message(
+                MyException3,
+                "my exception short circuit",
+                conn.execute, "SELECT 'ERROR FOUR' FROM I_DONT_EXIST"
+            )
+            eq_(patched.call_count, 1)
+
+
 
     @testing.fails_on('firebird', 'Data type unknown')
     def test_execute_events(self):