- rework the entire approach to #3076. As we need to catch all exceptions

in all cases unconditionally, the number of use cases that go beyond what
dbapi_error() is expecting has gone too far for an 0.9 release.
Additionally, the number of things we'd like to track is really a lot
more than the five arguments here, and ExecutionContext is really not
suitable as totally public API for this.   So restore dbapi_error
to its old version, deprecate, and build out handle_error instead.
This is a lot more extensible and doesn't get in the way of anything
compatibility-wise.

8 files changed, 517 insertions, 235 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index d3bce5cbe..0dca50688 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -19,20 +19,9 @@
         :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: feature, engine
-        :versions: 1.0.0
-
-        Added new attributes :attr:`.ExecutionContext.exception` and
-        :attr:`.ExecutionContext.is_disconnect` which are meaningful within
-        the :meth:`.ConnectionEvents.dbapi_error` handler to see both the
-        original DBAPI error as well as whether or not it represents
-        a disconnect.
+        Added new event :meth:`.ConnectionEvents.handle_error`, a more
+        fully featured and comprehensive replacement for
+        :meth:`.ConnectionEvents.dbapi_error`.
 
     .. change::
         :tags: bug, orm
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index b9c45642c..06fccd1dd 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -133,33 +133,6 @@ 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/doc/build/core/connections.rst b/doc/build/core/connections.rst
index ca584f012..248309a2e 100644
--- a/doc/build/core/connections.rst
+++ b/doc/build/core/connections.rst
@@ -568,6 +568,9 @@ Connection / Engine API
 .. autoclass:: Engine
    :members:
 
+.. autoclass:: sqlalchemy.engine.ExceptionContext
+   :members:
+
 .. autoclass:: NestedTransaction
     :members:
 
diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index fcb38b09c..9c6460858 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -54,6 +54,7 @@ from .interfaces import (
     Connectable,
     Dialect,
     ExecutionContext,
+    ExceptionContext,
 
     # backwards compat
     Compiled,
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 67772f131..6da41927f 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -12,8 +12,8 @@ from __future__ import with_statement
 
 import sys
 from .. import exc, util, log, interfaces
-from ..sql import expression, util as sql_util, schema, ddl
-from .interfaces import Connectable, Compiled
+from ..sql import util as sql_util
+from .interfaces import Connectable, ExceptionContext
 from .util import _distill_params
 import contextlib
 
@@ -1096,28 +1096,51 @@ class Connection(Connectable):
             should_wrap = isinstance(e, self.dialect.dbapi.Error) or \
                 (statement is not None and context is None)
 
+            if should_wrap:
+                sqlalchemy_exception = exc.DBAPIError.instance(
+                    statement,
+                    parameters,
+                    e,
+                    self.dialect.dbapi.Error,
+                    connection_invalidated=self._is_disconnect)
+            else:
+                sqlalchemy_exception = None
+
             newraise = None
-            if should_wrap and context:
-                if self._has_events or self.engine._has_events:
-                    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
 
+            if self._has_events or self.engine._has_events:
+                # legacy dbapi_error event
+                if should_wrap and context:
+                    self.dispatch.dbapi_error(self,
+                                                    cursor,
+                                                    statement,
+                                                    parameters,
+                                                    context,
+                                                    e)
+
+                # new handle_error event
+                ctx = ExceptionContextImpl(
+                    e, sqlalchemy_exception, self, cursor, statement,
+                    parameters, context, self._is_disconnect)
+
+                for fn in self.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 \
+                        self._is_disconnect != ctx.is_disconnect:
+                    sqlalchemy_exception.connection_invalidated = \
+                        self._is_disconnect = ctx.is_disconnect
 
+            if should_wrap and context:
                 context.handle_dbapi_exception(e)
 
             if not self._is_disconnect:
@@ -1129,16 +1152,11 @@ class Connection(Connectable):
                 util.raise_from_cause(newraise, exc_info)
             elif should_wrap:
                 util.raise_from_cause(
-                                    exc.DBAPIError.instance(
-                                        statement,
-                                        parameters,
-                                        e,
-                                        self.dialect.dbapi.Error,
-                                        connection_invalidated=self._is_disconnect),
+                                    sqlalchemy_exception,
                                     exc_info
                                 )
-
-            util.reraise(*exc_info)
+            else:
+                util.reraise(*exc_info)
 
         finally:
             del self._reentrant_error
@@ -1224,6 +1242,21 @@ class Connection(Connectable):
                             **kwargs).traverse_single(element)
 
 
+class ExceptionContextImpl(ExceptionContext):
+    """Implement the :class:`.ExceptionContext` interface."""
+
+    def __init__(self, exception, sqlalchemy_exception,
+                        connection, cursor, statement, parameters,
+                        context, is_disconnect):
+        self.connection = connection
+        self.sqlalchemy_exception = sqlalchemy_exception
+        self.original_exception = exception
+        self.execution_context = context
+        self.statement = statement
+        self.parameters = parameters
+        self.is_disconnect = is_disconnect
+
+
 class Transaction(object):
     """Represent a database transaction in progress.
 
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 230d00fc0..1807e4283 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -882,3 +882,105 @@ class Connectable(object):
 
     def _execute_clauseelement(self, elem, multiparams=None, params=None):
         raise NotImplementedError()
+
+class ExceptionContext(object):
+    """Encapsulate information about an error condition in progress.
+
+    This object exists solely to be passed to the
+    :meth:`.ConnectionEvents.handle_error` event, supporting an interface that
+    can be extended without backwards-incompatibility.
+
+    .. versionadded:: 0.9.7
+
+    """
+
+    connection = None
+    """The :class:`.Connection` in use during the exception.
+
+    This member is always present.
+
+    """
+
+    cursor = None
+    """The DBAPI cursor object.
+
+    May be None.
+
+    """
+
+    statement = None
+    """String SQL statement that was emitted directly to the DBAPI.
+
+    May be None.
+
+    """
+
+    parameters = None
+    """Parameter collection that was emitted directly to the DBAPI.
+
+    May be None.
+
+    """
+
+    original_exception = None
+    """The exception object which was caught.
+
+    This member is always present.
+
+    """
+
+    sqlalchemy_exception = None
+    """The :class:`sqlalchemy.exc.StatementError` which wraps the original,
+    and will be raised if exception handling is not circumvented by the event.
+
+    May be None, as not all exception types are wrapped by SQLAlchemy.
+    For DBAPI-level exceptions that subclass the dbapi's Error class, this
+    field will always be present.
+
+    """
+
+    chained_exception = None
+    """The exception that was returned by the previous handler in the
+    exception chain, if any.
+
+    If present, this exception will be the one ultimately raised by
+    SQLAlchemy unless a subsequent handler replaces it.
+
+    May be None.
+
+    """
+
+    execution_context = None
+    """The :class:`.ExecutionContext` corresponding to the execution
+    operation in progress.
+
+    This is present for statement execution operations, but not for
+    operations such as transaction begin/end.  It also is not present when
+    the exception was raised before the :class:`.ExecutionContext`
+    could be constructed.
+
+    Note that the :attr:`.ExceptionContext.statement` and
+    :attr:`.ExceptionContext.parameters` members may represent a
+    different value than that of the :class:`.ExecutionContext`,
+    potentially in the case where a
+    :meth:`.ConnectionEvents.before_cursor_execute` event or similar
+    modified the statement/parameters to be sent.
+
+    May be None.
+
+    """
+
+    is_disconnect = None
+    """Represent whether the exception as occurred represents a "disconnect"
+    condition.
+
+    This flag will always be True or False within the scope of the
+    :meth:`.ConnectionEvents.handle_error` handler.
+
+    SQLAlchemy will defer to this flag in order to determine whether or not
+    the connection should be invalidated subsequently.    That is, by
+    assigning to this flag, a "disconnect" event which then results in
+    a connection and pool invalidation can be invoked or prevented by
+    changing this flag.
+
+    """
diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py
index d3fe80b76..e4bc48615 100644
--- a/lib/sqlalchemy/events.py
+++ b/lib/sqlalchemy/events.py
@@ -494,10 +494,10 @@ class ConnectionEvents(event.Events):
                 fn = wrap_before_cursor_execute
         elif retval and \
             identifier not in ('before_execute',
-                                    'before_cursor_execute', 'dbapi_error'):
+                                    'before_cursor_execute', 'handle_error'):
             raise exc.ArgumentError(
                     "Only the 'before_execute', "
-                    "'before_cursor_execute' and 'dbapi_error' engine "
+                    "'before_cursor_execute' and 'handle_error' engine "
                     "event listeners accept the 'retval=True' "
                     "argument.")
         event_key.with_wrapper(fn).base_listen()
@@ -611,16 +611,72 @@ class ConnectionEvents(event.Events):
 
         This event is called with the DBAPI exception instance
         received from the DBAPI itself, *before* SQLAlchemy wraps the
-        exception with its own exception wrappers, and before any
+        exception with it's own exception wrappers, and before any
         other operations are performed on the DBAPI cursor; the
         existing transaction remains in effect as well as any state
         on the cursor.
 
-        The use cases supported by this hook include:
+        The use case here is to inject low-level exception handling
+        into an :class:`.Engine`, typically for logging and
+        debugging purposes.
+
+        .. warning::
+
+            Code should **not** modify
+            any state or throw any exceptions here as this will
+            interfere with SQLAlchemy's cleanup and error handling
+            routines.  For exception modification, please refer to the
+            new :meth:`.ConnectionEvents.handle_error` event.
+
+        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.
+
+        :param conn: :class:`.Connection` object
+        :param cursor: DBAPI cursor object
+        :param statement: string SQL statement
+        :param parameters: Dictionary, tuple, or list of parameters being
+         passed to the ``execute()`` or ``executemany()`` method of the
+         DBAPI ``cursor``.  In some cases may be ``None``.
+        :param context: :class:`.ExecutionContext` object in use.  May
+         be ``None``.
+        :param exception: The **unwrapped** exception emitted directly from the
+         DBAPI.  The class here is specific to the DBAPI module in use.
+
+        .. deprecated:: 0.9.7 - replaced by
+            :meth:`.ConnectionEvents.handle_error`
+
+        """
+
+    def handle_error(self, exception_context):
+        """Intercept all exceptions processed by the :class:`.Connection`.
+
+        This includes all exceptions emitted by the DBAPI as well as
+        within SQLAlchemy's statement invocation process, including
+        encoding errors and other statement validation errors.  Other areas
+        in which the event is invoked include transaction begin and end,
+        result row fetching, cursor creation.
+
+        Note that :meth:`.handle_error` may support new kinds of exceptions
+        and new calling scenarios at *any time*.  Code which uses this
+        event must expect new calling patterns to be present in minor
+        releases.
+
+        To support the wide variety of members that correspond to an exception,
+        as well as to allow extensibility of the event without backwards
+        incompatibility, the sole argument received is an instance of
+        :class:`.ExceptionContext`.   This object contains data members
+        representing detail about the exception.
+
+        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)
+        * exception re-writing
 
         The hook is called while the cursor from the failed operation
         (if any) is still open and accessible.   Special cleanup operations
@@ -630,10 +686,6 @@ class ConnectionEvents(event.Events):
         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
@@ -641,28 +693,26 @@ class ConnectionEvents(event.Events):
         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):
+            @event.listens_for(Engine, "handle_error")
+            def handle_exception(context):
+                if isinstance(context.original_exception,
+                    psycopg2.OperationalError) and \\
+                    "failed" in str(context.original_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
+        next handler.   The "chained" exception is available using
+        :attr:`.ExceptionContext.chained_exception`::
+
+            @event.listens_for(Engine, "handle_error", retval=True)
+            def handle_exception(context):
+                if context.chained_exception is not None and \\
+                    "special" in context.chained_exception.message:
+                    return MySpecialException("failed",
+                        cause=context.chained_exception)
 
         Handlers that return ``None`` may remain within this chain; the
         last non-``None`` return value is the one that continues to be
@@ -676,22 +726,11 @@ class ConnectionEvents(event.Events):
         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
-        :param statement: string SQL statement
-        :param parameters: Dictionary, tuple, or list of parameters being
-         passed to the ``execute()`` or ``executemany()`` method of the
-         DBAPI ``cursor``.  In some cases may be ``None``.
-        :param context: :class:`.ExecutionContext` object in use.  May
-         be ``None``.
-        :param exception: The **unwrapped** exception emitted directly from the
-         DBAPI.  The class here is specific to the DBAPI module in use.
+        :param context: an :class:`.ExceptionContext` object.  See this
+         class for details on all available members.
 
-        .. versionadded:: 0.7.7
+        .. versionadded:: 0.9.7 Added the
+            :meth:`.ConnectionEvents.handle_error` hook.
 
         """
 
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 7e7126fce..d511dce53 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1115,131 +1115,6 @@ class EngineEventsTest(fixtures.TestBase):
         e1.execute(select([1]).compile(dialect=e1.dialect))
         e1._execute_compiled(select([1]).compile(dialect=e1.dialect), (), {})
 
-    def test_exception_event(self):
-        engine = engines.testing_engine()
-        canary = []
-
-        @event.listens_for(engine, 'dbapi_error')
-        def err(conn, cursor, stmt, parameters, context, exception):
-            canary.append((stmt, parameters, exception))
-
-        conn = engine.connect()
-        try:
-            conn.execute("SELECT FOO FROM I_DONT_EXIST")
-            assert False
-        except tsa.exc.DBAPIError as e:
-            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)
 
 
 
@@ -1516,6 +1391,273 @@ class EngineEventsTest(fixtures.TestBase):
                        'prepare_twophase', 'commit_twophase']
         )
 
+class HandleErrorTest(fixtures.TestBase):
+    __requires__ = 'ad_hoc_engines',
+    __backend__ = True
+
+    def tearDown(self):
+        Engine.dispatch._clear()
+        Engine._has_events = False
+
+    def test_legacy_dbapi_error(self):
+        engine = engines.testing_engine()
+        canary = Mock()
+
+        event.listen(engine, "dbapi_error", canary)
+
+        with engine.connect() as conn:
+            try:
+                conn.execute("SELECT FOO FROM I_DONT_EXIST")
+                assert False
+            except tsa.exc.DBAPIError as e:
+                eq_(canary.mock_calls[0][1][5], e.orig)
+                eq_(canary.mock_calls[0][1][2], "SELECT FOO FROM I_DONT_EXIST")
+
+    def test_legacy_dbapi_error_no_ad_hoc_context(self):
+        engine = engines.testing_engine()
+
+        listener = Mock(return_value=None)
+        event.listen(engine, 'dbapi_error', listener)
+
+        nope = Exception("nope")
+        class MyType(TypeDecorator):
+            impl = Integer
+            def process_bind_param(self, value, dialect):
+                raise nope
+
+        with engine.connect() as conn:
+            assert_raises_message(
+                tsa.exc.StatementError,
+                r"nope \(original cause: Exception: nope\) u?'SELECT 1 ",
+                conn.execute,
+                    select([1]).where(
+                            column('foo') == literal('bar', MyType()))
+            )
+        # no legacy event
+        eq_(listener.mock_calls, [])
+
+    def test_legacy_dbapi_error_non_dbapi_error(self):
+        engine = engines.testing_engine()
+
+        listener = Mock(return_value=None)
+        event.listen(engine, 'dbapi_error', listener)
+
+        nope = TypeError("I'm not a DBAPI error")
+        with engine.connect() as c:
+            c.connection.cursor = Mock(
+                    return_value=Mock(
+                        execute=Mock(
+                                side_effect=nope
+                        ))
+                    )
+
+            assert_raises_message(
+                TypeError,
+                "I'm not a DBAPI error",
+                c.execute, "select "
+            )
+        # no legacy event
+        eq_(listener.mock_calls, [])
+
+
+    def test_handle_error(self):
+        engine = engines.testing_engine()
+        canary = Mock(return_value=None)
+
+        event.listen(engine, "handle_error", canary)
+
+        with engine.connect() as conn:
+            try:
+                conn.execute("SELECT FOO FROM I_DONT_EXIST")
+                assert False
+            except tsa.exc.DBAPIError as e:
+                ctx = canary.mock_calls[0][1][0]
+
+                eq_(ctx.original_exception, e.orig)
+                is_(ctx.sqlalchemy_exception, e)
+                eq_(ctx.statement, "SELECT FOO FROM I_DONT_EXIST")
+
+    def test_exception_event_reraise(self):
+        engine = engines.testing_engine()
+
+        class MyException(Exception):
+            pass
+
+        @event.listens_for(engine, 'handle_error', retval=True)
+        def err(context):
+            stmt = context.statement
+            exception = context.original_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, 'handle_error', retval=True)
+        def err1(context):
+            stmt = context.statement
+
+            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, 'handle_error', retval=True)
+        def err2(context):
+            stmt = context.statement
+            if ("ERROR ONE" in str(stmt) or "ERROR FOUR" in str(stmt)) \
+                    and isinstance(context.chained_exception, MyException1):
+                raise MyException2("my exception chained")
+            elif "ERROR TWO" in str(stmt):
+                return context.chained_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)
+
+    def test_exception_event_ad_hoc_context(self):
+        """test that handle_error is called with a context in
+        cases where _handle_dbapi_error() is normally called without
+        any context.
+
+        """
+
+        engine = engines.testing_engine()
+
+        listener = Mock(return_value=None)
+        event.listen(engine, 'handle_error', listener)
+
+        nope = Exception("nope")
+        class MyType(TypeDecorator):
+            impl = Integer
+            def process_bind_param(self, value, dialect):
+                raise nope
+
+        with engine.connect() as conn:
+            assert_raises_message(
+                tsa.exc.StatementError,
+                r"nope \(original cause: Exception: nope\) u?'SELECT 1 ",
+                conn.execute,
+                    select([1]).where(
+                            column('foo') == literal('bar', MyType()))
+            )
+
+        ctx = listener.mock_calls[0][1][0]
+        assert ctx.statement.startswith("SELECT 1 ")
+        is_(ctx.is_disconnect, False)
+        is_(ctx.original_exception, nope)
+
+    def test_exception_event_non_dbapi_error(self):
+        """test that dbapi_error is called with a context in
+        cases where DBAPI raises an exception that is not a DBAPI
+        exception, e.g. internal errors or encoding problems.
+
+        """
+        engine = engines.testing_engine()
+
+        listener = Mock(return_value=None)
+        event.listen(engine, 'handle_error', listener)
+
+        nope = TypeError("I'm not a DBAPI error")
+        with engine.connect() as c:
+            c.connection.cursor = Mock(
+                    return_value=Mock(
+                        execute=Mock(
+                                side_effect=nope
+                        ))
+                    )
+
+            assert_raises_message(
+                TypeError,
+                "I'm not a DBAPI error",
+                c.execute, "select "
+            )
+        ctx = listener.mock_calls[0][1][0]
+        eq_(ctx.statement, "select ")
+        is_(ctx.is_disconnect, False)
+        is_(ctx.original_exception, nope)
 
 class ProxyConnectionTest(fixtures.TestBase):
     """These are the same tests as EngineEventsTest, except using