- TypeDecorator is present in the "sqlalchemy" import space.
+ - Non-DBAPI errors which occur in the scope of an `execute()`
+ call are now wrapped in sqlalchemy.exc.StatementError,
+ and the text of the SQL statement and repr() of params
+ is included. This makes it easier to identify statement
+ executions which fail before the DBAPI becomes
+ involved. [ticket:2015]
+
-sqlite
- SQLite dialect now uses `NullPool` for file-based databases
[ticket:1921]
have = rs.rowcount > 0
rs.close()
return have
- except exc.SQLError, e:
+ except exc.DBAPIError, e:
if self._extract_error_code(e.orig) == 1146:
return False
raise
rp = None
try:
rp = connection.execute(st)
- except exc.SQLError, e:
+ except exc.DBAPIError, e:
if self._extract_error_code(e.orig) == 1146:
raise exc.NoSuchTableError(full_name)
else:
try:
try:
rp = connection.execute(st)
- except exc.SQLError, e:
+ except exc.DBAPIError, e:
if self._extract_error_code(e.orig) == 1146:
raise exc.NoSuchTableError(full_name)
else:
"""Execute a schema.DDL object."""
dialect = self.dialect
+
+ compiled = ddl.compile(dialect=dialect)
return self._execute_context(
dialect,
dialect.execution_ctx_cls._init_ddl,
- None,
+ compiled,
None,
- ddl.compile(dialect=dialect)
+ compiled
)
def _execute_clauseelement(self, elem, multiparams, params):
return self._execute_context(
dialect,
dialect.execution_ctx_cls._init_compiled,
- None,
+ compiled_sql,
params,
compiled_sql, params
)
return self._execute_context(
dialect,
dialect.execution_ctx_cls._init_compiled,
- None,
+ compiled,
parameters,
compiled, parameters
)
_after_cursor_execute = None
def _execute_context(self, dialect, constructor,
- statement, parameters, *args):
+ statement, parameters,
+ *args):
"""Create an :class:`.ExecutionContext` and execute, returning
a :class:`.ResultProxy`."""
context = constructor(dialect, self, conn, *args)
except Exception, e:
self._handle_dbapi_exception(e,
- statement, parameters,
+ str(statement), parameters,
None, None)
raise
context):
if getattr(self, '_reentrant_error', False):
# Py3K
- #raise exc.DBAPIError.instance(statement, parameters, e) from e
+ #raise exc.DBAPIError.instance(statement, parameters, e,
+ # self.dialect.dbapi.Error) from e
# Py2K
- raise exc.DBAPIError.instance(statement, parameters, e), \
+ raise exc.DBAPIError.instance(statement,
+ parameters,
+ e,
+ self.dialect.dbapi.Error), \
None, sys.exc_info()[2]
# end Py2K
self._reentrant_error = True
try:
- if not isinstance(e, self.dialect.dbapi.Error):
+ # non-DBAPI error - if we already got a context,
+ # or theres no string statement, don't wrap it
+ if not isinstance(e, self.dialect.dbapi.Error) and \
+ (statement is None or context is not None):
return
if context:
context.handle_dbapi_exception(e)
- is_disconnect = self.dialect.is_disconnect(e, self.__connection, cursor)
+ is_disconnect = isinstance(e, self.dialect.dbapi.Error) and \
+ self.dialect.is_disconnect(e, self.__connection, cursor)
if is_disconnect:
self.invalidate(e)
self.engine.dispose()
# statement,
# parameters,
# e,
+ # self.dialect.dbapi.Error,
# connection_invalidated=is_disconnect) \
# from e
# Py2K
statement,
parameters,
e,
+ self.dialect.dbapi.Error,
connection_invalidated=is_disconnect), \
None, sys.exc_info()[2]
# end Py2K
self.compiled = compiled
if not compiled.can_execute:
- raise exc.ArgumentError("Not an executable clause: %s" % compiled)
+ raise exc.ArgumentError("Not an executable clause")
self.execution_options = compiled.statement._execution_options
if connection._execution_options:
return dialect.connect(*cargs, **cparams)
except Exception, e:
# Py3K
- #raise exc.DBAPIError.instance(None, None, e) from e
+ #raise exc.DBAPIError.instance(None, None,
+ # dialect.dbapi.Error, e) from e
# Py2K
import sys
- raise exc.DBAPIError.instance(None, None, e), None, sys.exc_info()[2]
+ raise exc.DBAPIError.instance(
+ None, None, e, dialect.dbapi.Error), \
+ None, sys.exc_info()[2]
# end Py2K
creator = kwargs.pop('creator', connect)
# Moved to orm.exc; compatability definition installed by orm import until 0.6
UnmappedColumnError = None
-class DBAPIError(SQLAlchemyError):
+class StatementError(SQLAlchemyError):
+ """An error occured during execution of a SQL statement.
+
+ :class:`.StatementError` wraps the exception raised
+ during execution, and features :attr:`.statement`
+ and :attr:`.params` attributes which supply context regarding
+ the specifics of the statement which had an issue.
+
+ The wrapped exception object is available in
+ the :attr:`.orig` attribute.
+
+ """
+
+ def __init__(self, message, statement, params, orig):
+ SQLAlchemyError.__init__(self, message)
+ self.statement = statement
+ self.params = params
+ self.orig = orig
+
+ def __str__(self):
+ if isinstance(self.params, (list, tuple)) and \
+ len(self.params) > 10 and \
+ isinstance(self.params[0], (list, dict, tuple)):
+ return ' '.join((SQLAlchemyError.__str__(self),
+ repr(self.statement),
+ repr(self.params[:2]),
+ '... and a total of %i bound parameter sets' % len(self.params)))
+ return ' '.join((SQLAlchemyError.__str__(self),
+ repr(self.statement), repr(self.params)))
+
+class DBAPIError(StatementError):
"""Raised when the execution of a database operation fails.
``DBAPIError`` wraps exceptions raised by the DB-API underlying the
that there is no guarantee that different DB-API implementations will
raise the same exception type for any given error condition.
- If the error-raising operation occured in the execution of a SQL
- statement, that statement and its parameters will be available on
- the exception object in the ``statement`` and ``params`` attributes.
+ :class:`.DBAPIError` features :attr:`.statement`
+ and :attr:`.params` attributes which supply context regarding
+ the specifics of the statement which had an issue, for the
+ typical case when the error was raised within the context of
+ emitting a SQL statement.
- The wrapped exception object is available in the ``orig`` attribute.
+ The wrapped exception object is available in the :attr:`.orig` attribute.
Its type and properties are DB-API implementation specific.
"""
@classmethod
- def instance(cls, statement, params, orig, connection_invalidated=False):
+ def instance(cls, statement, params,
+ orig,
+ dbapi_base_err,
+ connection_invalidated=False):
# Don't ever wrap these, just return them directly as if
# DBAPIError didn't exist.
if isinstance(orig, (KeyboardInterrupt, SystemExit)):
return orig
if orig is not None:
+ # not a DBAPI error, statement is present.
+ # raise a StatementError
+ if not isinstance(orig, dbapi_base_err) and statement:
+ return StatementError(str(orig), statement, params, orig)
+
name, glob = orig.__class__.__name__, globals()
if name in glob and issubclass(glob[name], DBAPIError):
cls = glob[name]
raise
except Exception, e:
text = 'Error in str() of DB-API-generated exception: ' + str(e)
- SQLAlchemyError.__init__(
- self, '(%s) %s' % (orig.__class__.__name__, text))
- self.statement = statement
- self.params = params
- self.orig = orig
+ StatementError.__init__(
+ self,
+ '(%s) %s' % (orig.__class__.__name__, text),
+ statement,
+ params,
+ orig
+ )
self.connection_invalidated = connection_invalidated
- def __str__(self):
- if isinstance(self.params, (list, tuple)) and len(self.params) > 10 and isinstance(self.params[0], (list, dict, tuple)):
- return ' '.join((SQLAlchemyError.__str__(self),
- repr(self.statement),
- repr(self.params[:2]),
- '... and a total of %i bound parameter sets' % len(self.params)))
- return ' '.join((SQLAlchemyError.__str__(self),
- repr(self.statement), repr(self.params)))
-
-
-# As of 0.4, SQLError is now DBAPIError.
-# SQLError alias will be removed in 0.6.
-SQLError = DBAPIError
class InterfaceError(DBAPIError):
"""Wraps a DB-API InterfaceError."""
def test_db_error_normal(self):
try:
raise sa_exceptions.DBAPIError.instance('', [],
- OperationalError())
+ OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError:
self.assert_(True)
def test_tostring(self):
try:
raise sa_exceptions.DBAPIError.instance('this is a message'
- , None, OperationalError())
+ , None, OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc) \
== "(OperationalError) 'this is a message' None"
,
{'a': 1, 'b': 2, 'c': 3, 'd': 4, 'e': 5, 'f': 6, 'g': 7, 'h':
8, 'i': 9, 'j': 10, 'k': 11,
- }, OperationalError())
+ }, OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc).startswith("(OperationalError) 'this is a "
"message' {")
def test_tostring_large_list(self):
try:
raise sa_exceptions.DBAPIError.instance('this is a message',
- [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11,], OperationalError())
+ [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11,],
+ OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc).startswith("(OperationalError) 'this is a "
"message' [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]")
raise sa_exceptions.DBAPIError.instance('this is a message',
[{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1},
{1: 1}, {1:1}, {1: 1}, {1: 1},],
- OperationalError())
+ OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc) \
== "(OperationalError) 'this is a message' [{1: 1}, "\
raise sa_exceptions.DBAPIError.instance('this is a message', [
{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1},
{1:1}, {1: 1}, {1: 1}, {1: 1},
- ], OperationalError())
+ ], OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc) \
== "(OperationalError) 'this is a message' [{1: 1}, "\
[
(1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ),
(1, ),
- ], OperationalError())
+ ], OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc) \
== "(OperationalError) 'this is a message' [(1,), "\
raise sa_exceptions.DBAPIError.instance('this is a message', [
(1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ),
(1, ), (1, ),
- ], OperationalError())
+ ], OperationalError(), DatabaseError)
except sa_exceptions.DBAPIError, exc:
assert str(exc) \
== "(OperationalError) 'this is a message' [(1,), "\
def test_db_error_busted_dbapi(self):
try:
raise sa_exceptions.DBAPIError.instance('', [],
- ProgrammingError())
+ ProgrammingError(), DatabaseError)
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('', [], OutOfSpec())
+ raise sa_exceptions.DBAPIError.instance('', [], OutOfSpec(),
+ DatabaseError)
except sa_exceptions.DBAPIError, e:
self.assert_(e.__class__ is sa_exceptions.DBAPIError)
except OutOfSpec:
self.assert_(False)
- # Make sure the DatabaseError recognition logic is limited to
- # subclasses of sqlalchemy.exceptions.DBAPIError
-
try:
raise sa_exceptions.DBAPIError.instance('', [],
- sa_exceptions.ArgumentError())
+ sa_exceptions.ArgumentError(), DatabaseError)
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('', [],
- KeyboardInterrupt())
+ KeyboardInterrupt(), DatabaseError)
except sa_exceptions.DBAPIError:
self.assert_(False)
except KeyboardInterrupt:
def test_db_error_system_exit(self):
try:
raise sa_exceptions.DBAPIError.instance('', [],
- SystemExit())
+ SystemExit(), DatabaseError)
except sa_exceptions.DBAPIError:
self.assert_(False)
except SystemExit:
enum_table.drop(checkfirst=True)
enum_table.create()
- assert_raises(exc.SQLError, enum_table.insert().execute,
+ assert_raises(exc.DBAPIError, enum_table.insert().execute,
e1=None, e2=None, e3=None, e4=None)
- assert_raises(exc.InvalidRequestError, enum_table.insert().execute,
+ assert_raises(exc.StatementError, enum_table.insert().execute,
e1='c', e2='c', e2generic='c', e3='c',
e4='c', e5='c', e5generic='c', e6='c')
:
try:
con.execute(ddl)
- except exc.SQLError, e:
+ except exc.DBAPIError, e:
if not 'already exists' in str(e):
raise e
con.execute('CREATE TABLE testtable (question integer, answer '
meta.drop_all()
def test_string_dates_raise(self):
- assert_raises(TypeError, testing.db.execute,
+ assert_raises(exc.StatementError, testing.db.execute,
select([1]).where(bindparam('date', type_=Date)),
date=str(datetime.date(2007, 10, 30)))
-from test.lib.testing import eq_, assert_raises
+from test.lib.testing import eq_, assert_raises, assert_raises_message
import re
from sqlalchemy.interfaces import ConnectionProxy
from sqlalchemy import MetaData, Integer, String, INT, VARCHAR, func, \
- bindparam, select, event
+ bindparam, select, event, TypeDecorator
+from sqlalchemy.sql import column, literal
from test.lib.schema import Table, Column
import sqlalchemy as tsa
from test.lib import TestBase, testing, engines
'horse'), (4, 'sally')]
conn.execute('delete from users')
- def test_exception_wrapping(self):
+ def test_exception_wrapping_dbapi(self):
for conn in testing.db, testing.db.connect():
- try:
- conn.execute('osdjafioajwoejoasfjdoifjowejfoawejqoijwef'
- )
- assert False
- except tsa.exc.DBAPIError:
- assert True
+ assert_raises_message(
+ tsa.exc.DBAPIError,
+ r"not_a_valid_statement",
+ conn.execute, 'not_a_valid_statement'
+ )
+
+ def test_exception_wrapping_non_dbapi_statement(self):
+ class MyType(TypeDecorator):
+ impl = Integer
+ def process_bind_param(self, value, dialect):
+ raise Exception("nope")
+
+ for conn in testing.db, testing.db.connect():
+ assert_raises_message(
+ tsa.exc.StatementError,
+ "nope 'SELECT 1 ",
+ conn.execute,
+ select([1]).\
+ where(
+ column('foo') == literal('bar', MyType())
+ )
+ )
def test_empty_insert(self):
"""test that execute() interprets [] as a list with no params"""
-from test.lib.testing import eq_, assert_raises
+from test.lib.testing import eq_, assert_raises, assert_raises_message
import time
import weakref
from sqlalchemy import select, MetaData, Integer, String, pool
assert not conn.closed
assert conn.invalidated
assert trans.is_active
- try:
- conn.execute(select([1]))
- assert False
- except tsa.exc.InvalidRequestError, e:
- assert str(e) \
- == "Can't reconnect until invalid transaction is "\
- "rolled back"
+ assert_raises_message(
+ tsa.exc.StatementError,
+ "Can't reconnect until invalid transaction is rolled back",
+ conn.execute, select([1])
+ )
assert trans.is_active
try:
trans.commit()
assert not conn.closed
assert conn.invalidated
assert trans.is_active
- try:
- conn.execute(select([1]))
- assert False
- except tsa.exc.InvalidRequestError, e:
- assert str(e) \
- == "Can't reconnect until invalid transaction is "\
- "rolled back"
+ assert_raises_message(
+ tsa.exc.StatementError,
+ "Can't reconnect until invalid transaction is "\
+ "rolled back",
+ conn.execute, select([1])
+ )
assert trans.is_active
try:
trans.commit()
metadata.create_all()
foo.insert().execute(id=1,x=9,y=5)
- assert_raises(exc.SQLError, foo.insert().execute, id=2,x=5,y=9)
+ assert_raises(exc.DBAPIError, foo.insert().execute, id=2,x=5,y=9)
bar.insert().execute(id=1,x=10)
- assert_raises(exc.SQLError, bar.insert().execute, id=2,x=5)
+ assert_raises(exc.DBAPIError, bar.insert().execute, id=2,x=5)
def test_unique_constraint(self):
foo = Table('foo', metadata,
foo.insert().execute(id=2, value='value2')
bar.insert().execute(id=1, value='a', value2='a')
bar.insert().execute(id=2, value='a', value2='b')
- assert_raises(exc.SQLError, foo.insert().execute, id=3, value='value1')
- assert_raises(exc.SQLError, bar.insert().execute, id=3, value='a', value2='b')
+ assert_raises(exc.DBAPIError, foo.insert().execute, id=3, value='value1')
+ assert_raises(exc.DBAPIError, bar.insert().execute, id=3, value='a', value2='b')
def test_index_create(self):
employees = Table('employees', metadata,
12, today, 'py')])
def test_missing_many_param(self):
- assert_raises_message(exc.InvalidRequestError,
+ assert_raises_message(exc.StatementError,
"A value is required for bind parameter 'col7', in parameter group 1",
t.insert().execute,
{'col4':7, 'col7':12, 'col8':19},
nonai.insert().execute(data='row 1')
nonai.insert().execute(data='row 2')
assert False
- except sa.exc.SQLError, e:
+ except sa.exc.DBAPIError, e:
assert True
nonai.insert().execute(id=1, data='row 1')
def test_insert_heterogeneous_params(self):
"""test that executemany parameters are asserted to match the parameter set of the first."""
- assert_raises_message(exc.InvalidRequestError,
- "A value is required for bind parameter 'user_name', in parameter group 2",
+ assert_raises_message(exc.StatementError,
+ "A value is required for bind parameter 'user_name', in parameter group 2 'INSERT INTO query_users",
users.insert().execute,
{'user_id':7, 'user_name':'jack'},
{'user_id':8, 'user_name':'ed'},
def test_cant_execute_join(self):
try:
users.join(addresses).execute()
- except exc.ArgumentError, e:
- assert str(e).startswith('Not an executable clause: ')
+ except exc.StatementError, e:
+ assert str(e).startswith('Not an executable clause ')
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
