"""
uniform interface for accessing the data in an `sqlalchemy.exc.IntegrityError`
"""
import pkg_resources
import sqlalchemy
from fictive.errors import ErrorData
[docs]class IntegrityErrorData(ErrorData):
"""
Records information assocated with an `sqlalchemy.exc.IntegrityError`
E.g.::
>>> first = Model(value=1)
>>> session.add(first)
>>> second = Model(value=1)
>>> session.add(second)
>>> try:
... session.commit()
>>> except sqlalchemy.exc.IntegrityError as integrity_error:
... data = IntegrityErrorData(integrity_error)
>>> data.code
'UniqueViolation'
>>> assert data.fields
{'value': '1'}
>>> assert data.table
'model'
"""
_ENTRY_POINT_GROUP = 'fictive.sqlalchemy.ext.errors.IntegrityErrorData'
#: The original error class apropriate for this ErrorData class
ERROR_CLASS = sqlalchemy.exc.IntegrityError
@classmethod
def _best_match(cls, error, error_data_classes):
def _is_instance(instance):
def _is_instance(klass):
return isinstance(instance, klass.ERROR_CLASS)
return _is_instance
def _matching_degree(instance):
def _matching_degree(klass):
instance_mro_set = set(type(instance).mro())
klass_mro_set = set(klass.ERROR_CLASS.mro())
return len(instance_mro_set - klass_mro_set)
return _matching_degree
try:
return sorted(
filter(_is_instance(error.orig), error_data_classes),
key=_matching_degree(error.orig)
)[0]
except IndexError:
try:
return sorted(
filter(_is_instance(error), error_data_classes),
key=_matching_degree(error)
)[0]
except IndexError:
raise TypeError(
f"{type(error).__name__} object {repr(error)}"
" is not a recognized integrity error"
)
def __new__(cls, integrity_error):
group = cls._ENTRY_POINT_GROUP
registered_classes = [
entry.load() for entry in pkg_resources.iter_entry_points(group)]
subclasses = list(
filter(lambda rc: issubclass(rc, cls), registered_classes))
best_subclass = cls._best_match(integrity_error, subclasses)
return super(IntegrityErrorData, cls).__new__(best_subclass)
[docs] def __init__(self, integrity_error: sqlalchemy.exc.IntegrityError):
# pylint: disable=super-init-not-called
"""
:param: integrity_error:
the `sqlalchemy.exc.IntegrityError` that contains a
databasedriver error in its `.orig` attribute
:type integrity_error: `sqlalchemy.exc.IntegrityError`
"""
error_match = isinstance(integrity_error, self.ERROR_CLASS)
orig_match = isinstance(
getattr(integrity_error, 'orig', None), self.ERROR_CLASS)
if not orig_match and not error_match:
raise TypeError(
f"expected a `{self.ERROR_CLASS.__name__}` instance")
self.integrity_error = integrity_error
@property
def code(self):
"""
use the class name as the default `code`
"""
return type(self.integrity_error).__name__
@property
def fields(self):
"""
table columns and values that caused the error
"""
return None
@property
def message(self):
"""
use the error itself as the default `message`
"""
return str(self.integrity_error)
@property
def table(self):
"""
the table that caused the error
"""
return None