--- /dev/null
+.. change::
+ :tags: feature, misc
+
+ Added a new errors section to the documentation with background
+ about common error messages. Selected exceptions within SQLAlchemy
+ will include a link in their string output to the relevant section
+ within this page.
"""
+from .util import compat
+
class SQLAlchemyError(Exception):
"""Generic error class."""
+ code = None
+
+ def __init__(self, *arg, **kw):
+ code = kw.pop('code', None)
+ if code is not None:
+ self.code = code
+ super(SQLAlchemyError, self).__init__(*arg, **kw)
+
+ def _code_str(self):
+ if not self.code:
+ return ""
+ else:
+ return (
+ "(Background on this error at: "
+ "http://sqlalche.me/e/%s)" % (self.code, )
+ )
+
+ def _message(self):
+ # get string representation just like Exception.__str__(self),
+ # but also support if the string has non-ascii chars
+ if len(self.args) == 1:
+ return compat.text_type(self.args[0])
+ else:
+ return compat.text_type(self.args)
+
+ def __str__(self):
+ message = self._message()
+
+ if self.code:
+ message = (
+ "%s %s" % (message, self._code_str())
+ )
+
+ return message
+
+ def __unicode__(self):
+ return self.__str__()
+
class ArgumentError(SQLAlchemyError):
"""Raised when an invalid or conflicting function argument is supplied.
see :ref:`use_alter`.
"""
- def __init__(self, message, cycles, edges, msg=None):
+ def __init__(self, message, cycles, edges, msg=None, code=None):
if msg is None:
message += " (%s)" % ", ".join(repr(s) for s in cycles)
else:
message = msg
- SQLAlchemyError.__init__(self, message)
+ SQLAlchemyError.__init__(self, message, code=code)
self.cycles = cycles
self.edges = edges
"""
invalidate_pool = True
+
class TimeoutError(SQLAlchemyError):
"""Raised when a connection pool times out on getting a connection."""
orig = None
"""The DBAPI exception object."""
- def __init__(self, message, statement, params, orig):
- SQLAlchemyError.__init__(self, message)
+ def __init__(self, message, statement, params, orig, code=None):
+ SQLAlchemyError.__init__(self, message, code=code)
self.statement = statement
self.params = params
self.orig = orig
def __str__(self):
from sqlalchemy.sql import util
- details = [SQLAlchemyError.__str__(self)]
+ details = [self._message()]
if self.statement:
details.append("[SQL: %r]" % self.statement)
if self.params:
params_repr = util._repr_params(self.params, 10)
details.append("[parameters: %r]" % params_repr)
+ code_str = self._code_str()
+ if code_str:
+ details.append(code_str)
return ' '.join([
"(%s)" % det for det in self.detail
] + details)
- def __unicode__(self):
- return self.__str__()
-
class DBAPIError(StatementError):
"""Raised when the execution of a database operation fails.
"""
+ code = 'dbapi'
+
@classmethod
def instance(cls, statement, params,
orig, dbapi_base_err,
if orig is not None:
# not a DBAPI error, statement is present.
# raise a StatementError
- if not isinstance(orig, dbapi_base_err) and statement:
+ if isinstance(orig, SQLAlchemyError) and statement:
+ return StatementError(
+ "(%s.%s) %s" %
+ (orig.__class__.__module__, orig.__class__.__name__,
+ orig.args[0]),
+ statement, params, orig, code=orig.code
+ )
+ elif not isinstance(orig, dbapi_base_err) and statement:
return StatementError(
"(%s.%s) %s" %
(orig.__class__.__module__, orig.__class__.__name__,
cls = glob[name]
break
- return cls(statement, params, orig, connection_invalidated)
+ return cls(statement, params, orig, connection_invalidated,
+ code=cls.code)
def __reduce__(self):
return self.__class__, (self.statement, self.params,
self.orig, self.connection_invalidated)
- def __init__(self, statement, params, orig, connection_invalidated=False):
+ def __init__(self, statement, params, orig, connection_invalidated=False,
+ code=None):
try:
text = str(orig)
except Exception as e:
orig.__class__.__module__, orig.__class__.__name__, text, ),
statement,
params,
- orig
+ orig, code=code
)
self.connection_invalidated = connection_invalidated
class InterfaceError(DBAPIError):
"""Wraps a DB-API InterfaceError."""
+ code = "rvf5"
+
class DatabaseError(DBAPIError):
"""Wraps a DB-API DatabaseError."""
+ code = "4xp6"
+
class DataError(DatabaseError):
"""Wraps a DB-API DataError."""
+ code = "9h9h"
+
class OperationalError(DatabaseError):
"""Wraps a DB-API OperationalError."""
+ code = "e3q8"
+
class IntegrityError(DatabaseError):
"""Wraps a DB-API IntegrityError."""
+ code = "gkpj"
+
class InternalError(DatabaseError):
"""Wraps a DB-API InternalError."""
+ code = "2j85"
+
class ProgrammingError(DatabaseError):
"""Wraps a DB-API ProgrammingError."""
+ code = "f405"
+
class NotSupportedError(DatabaseError):
"""Wraps a DB-API NotSupportedError."""
+ code = "tw8g"
# Warnings
"""An attempt to access unloaded attributes on a
mapped instance that is detached."""
+ code = "bhk3"
+
class UnmappedInstanceError(UnmappedError):
"""An mapping operation was requested for an unknown instance."""
raise exc.TimeoutError(
"QueuePool limit of size %d overflow %d reached, "
"connection timed out, timeout %d" %
- (self.size(), self.overflow(), self._timeout))
+ (self.size(), self.overflow(), self._timeout), code="3o7r")
if self._inc_overflow():
try:
if e is None:
raise exc.UnboundExecutionError(
"This Compiled object is not bound to any Engine "
- "or Connection.")
+ "or Connection.", code="2afi")
return e._execute_compiled(self, multiparams, params)
def scalar(self, *multiparams, **params):
raise exc.InvalidRequestError(
"A value is required for bind parameter %r, "
"in parameter group %d" %
- (bindparam.key, _group_number))
+ (bindparam.key, _group_number), code="cd3x")
else:
raise exc.InvalidRequestError(
"A value is required for bind parameter %r"
- % bindparam.key)
+ % bindparam.key, code="cd3x")
elif bindparam.callable:
pd[name] = bindparam.effective_value
raise exc.InvalidRequestError(
"A value is required for bind parameter %r, "
"in parameter group %d" %
- (bindparam.key, _group_number))
+ (bindparam.key, _group_number), code="cd3x")
else:
raise exc.InvalidRequestError(
"A value is required for bind parameter %r"
- % bindparam.key)
+ % bindparam.key, code="cd3x")
if bindparam.callable:
pd[self.bind_names[bindparam]] = bindparam.effective_value
+#! coding:utf-8
+
"""Tests exceptions and DB-API exception wrapping."""
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import eq_
from sqlalchemy.engine import default
+from sqlalchemy.util import u, compat
class Error(Exception):
eq_(
str(exc),
"(test.base.test_except.OperationalError) "
- "[SQL: 'this is a message']")
+ "[SQL: 'this is a message'] (Background on this error at: "
+ "http://sqlalche.me/e/e3q8)")
+
+ def test_statement_error_no_code(self):
+ try:
+ raise sa_exceptions.DBAPIError.instance(
+ 'select * from table', [{"x": 1}],
+ sa_exceptions.InvalidRequestError("hello"), DatabaseError)
+ except sa_exceptions.StatementError as err:
+ eq_(
+ str(err),
+ "(sqlalchemy.exc.InvalidRequestError) hello "
+ "[SQL: 'select * from table'] [parameters: [{'x': 1}]]"
+ )
+ eq_(err.args, ("(sqlalchemy.exc.InvalidRequestError) hello", ))
+
+ def test_statement_error_w_code(self):
+ try:
+ raise sa_exceptions.DBAPIError.instance(
+ 'select * from table', [{"x": 1}],
+ sa_exceptions.InvalidRequestError("hello", code="abcd"),
+ DatabaseError)
+ except sa_exceptions.StatementError as err:
+ eq_(
+ str(err),
+ "(sqlalchemy.exc.InvalidRequestError) hello "
+ "[SQL: 'select * from table'] [parameters: [{'x': 1}]] "
+ "(Background on this error at: http://sqlalche.me/e/abcd)"
+ )
+ eq_(err.args, ("(sqlalchemy.exc.InvalidRequestError) hello", ))
+
+ def test_wrap_multi_arg(self):
+ # this is not supported by the API but oslo_db is doing it
+ orig = sa_exceptions.DBAPIError(False, False, False)
+ orig.args = [2006, 'Test raise operational error']
+ eq_(
+ str(orig),
+ "(2006, 'Test raise operational error') "
+ "(Background on this error at: http://sqlalche.me/e/dbapi)"
+ )
+
+ def test_wrap_unicode_arg(self):
+ # this is not supported by the API but oslo_db is doing it
+ orig = sa_exceptions.DBAPIError(False, False, False)
+ orig.args = [u('méil')]
+ eq_(
+ compat.text_type(orig),
+ compat.u(
+ "méil (Background on this error at: "
+ "http://sqlalche.me/e/dbapi)")
+ )
+ eq_(orig.args, (u('méil'),))
def test_tostring_large_dict(self):
try:
'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(), DatabaseError)
+ OperationalError("sql error"), DatabaseError)
except sa_exceptions.DBAPIError as exc:
eq_(
str(exc),
- "(test.base.test_except.OperationalError) "
+ "(test.base.test_except.OperationalError) sql error "
"[SQL: 'this is a message'] [parameters: [{1: 1}, "
"{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: "
- "1}, {1: 1}, {1: 1}]]"
+ "1}, {1: 1}, {1: 1}]] (Background on this error at: "
+ "http://sqlalche.me/e/e3q8)"
+ )
+ eq_(
+ exc.args,
+ ("(test.base.test_except.OperationalError) sql error", )
)
try:
raise sa_exceptions.DBAPIError.instance('this is a message', [
"[SQL: 'this is a message'] [parameters: [{1: 1}, "
"{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, "
"{1: 1}, {1: 1} ... displaying 10 of 11 total "
- "bound parameter sets ... {1: 1}, {1: 1}]]"
+ "bound parameter sets ... {1: 1}, {1: 1}]] "
+ "(Background on this error at: http://sqlalche.me/e/e3q8)"
)
try:
raise sa_exceptions.DBAPIError.instance(
str(exc),
"(test.base.test_except.OperationalError) "
"[SQL: 'this is a message'] [parameters: [(1,), "
- "(1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,)]]")
+ "(1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,)]] "
+ "(Background on this error at: http://sqlalche.me/e/e3q8)")
try:
raise sa_exceptions.DBAPIError.instance('this is a message', [
(1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ),
"[SQL: 'this is a message'] [parameters: [(1,), "
"(1,), (1,), (1,), (1,), (1,), (1,), (1,) "
"... displaying 10 of 11 total bound "
- "parameter sets ... (1,), (1,)]]"
+ "parameter sets ... (1,), (1,)]] "
+ "(Background on this error at: http://sqlalche.me/e/e3q8)"
)
def test_db_error_busted_dbapi(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
