def _handle_dbapi_exception(self, e, statement, parameters, cursor, context):
if getattr(self, '_reentrant_error', False):
- raise exc.DBAPIError.instance_cls(e), (None, None, e), sys.exc_info()[2]
+ raise exc.DBAPIError.instance(statement, parameters, e), None, sys.exc_info()[2]
self._reentrant_error = True
try:
if not isinstance(e, self.dialect.dbapi.Error):
self._autorollback()
if self.__close_with_result:
self.close()
- raise exc.DBAPIError.instance_cls(e), (statement, parameters, e, is_disconnect), sys.exc_info()[2]
+ raise exc.DBAPIError.instance(statement, parameters, e, connection_invalidated=is_disconnect), None, sys.exc_info()[2]
finally:
del self._reentrant_error
"""
@classmethod
- def instance_cls(cls, orig):
+ def instance(cls, statement, params, orig, connection_invalidated=False):
# Don't ever wrap these, just return them directly as if
# DBAPIError didn't exist.
if isinstance(orig, (KeyboardInterrupt, SystemExit)):
- return type(orig)
+ return orig
if orig is not None:
name, glob = orig.__class__.__name__, globals()
if name in glob and issubclass(glob[name], DBAPIError):
cls = glob[name]
- return cls
+ return cls(statement, params, orig, connection_invalidated)
def __init__(self, statement, params, orig, connection_invalidated=False):
try:
class WrapTest(unittest.TestCase):
def test_db_error_normal(self):
try:
- raise sa_exceptions.DBAPIError.instance_cls(OperationalError()), \
- ('', [], OperationalError())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], OperationalError())
except sa_exceptions.DBAPIError:
self.assert_(True)
def test_db_error_busted_dbapi(self):
try:
- raise sa_exceptions.DBAPIError.instance_cls(ProgrammingError()), \
- ('', [], ProgrammingError())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], ProgrammingError())
except sa_exceptions.DBAPIError, e:
self.assert_(True)
self.assert_('Error in str() of DB-API' in e.args[0])
def test_db_error_noncompliant_dbapi(self):
try:
- raise sa_exceptions.DBAPIError.instance_cls(OutOfSpec()), \
- ('', [], OutOfSpec())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], OutOfSpec())
except sa_exceptions.DBAPIError, e:
self.assert_(e.__class__ is sa_exceptions.DBAPIError)
except OutOfSpec:
# Make sure the DatabaseError recognition logic is limited to
# subclasses of sqlalchemy.exceptions.DBAPIError
try:
- raise sa_exceptions.DBAPIError.instance_cls(sa_exceptions.ArgumentError()), \
- ('', [], sa_exceptions.ArgumentError())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], sa_exceptions.ArgumentError())
except sa_exceptions.DBAPIError, e:
self.assert_(e.__class__ is sa_exceptions.DBAPIError)
except sa_exceptions.ArgumentError:
def test_db_error_keyboard_interrupt(self):
try:
- raise sa_exceptions.DBAPIError.instance_cls(KeyboardInterrupt()), \
- ('', [], KeyboardInterrupt())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], KeyboardInterrupt())
except sa_exceptions.DBAPIError:
self.assert_(False)
except KeyboardInterrupt:
def test_db_error_system_exit(self):
try:
- raise sa_exceptions.DBAPIError.instance_cls(SystemExit()), \
- ('', [], SystemExit())
+ raise sa_exceptions.DBAPIError.instance(
+ '', [], SystemExit())
except sa_exceptions.DBAPIError:
self.assert_(False)
except SystemExit: