: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
.. _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
=======================
.. autoclass:: Engine
:members:
+.. autoclass:: sqlalchemy.engine.ExceptionContext
+ :members:
+
.. autoclass:: NestedTransaction
:members:
Connectable,
Dialect,
ExecutionContext,
+ ExceptionContext,
# backwards compat
Compiled,
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
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:
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
**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.
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.
+
+ """
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()
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
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
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
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.
"""
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)
'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
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
