x/deb-oslo.db
Code Issues Proposed changes

Merge "Raise DBReferenceError on foreign key violation"

Browse Source
This commit is contained in:
Jenkins 2014-07-25 13:44:10 +00:00 committed by Gerrit Code Review
commit 50b8d1e8a1
3 changed files with 131 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
oslo/db/exception.py
View File

@ -35,6 +35,17 @@ class DBDuplicateEntry(DBError):
super(DBDuplicateEntry, self).__init__(inner_exception)
class DBReferenceError(DBError):
"""Wraps an implementation specific exception."""
def __init__(self, table, constraint, key, key_table,
inner_exception=None):
self.table = table
self.constraint = constraint
self.key = key
self.key_table = key_table
super(DBReferenceError, self).__init__(inner_exception)
class DBDeadlock(DBError):
def __init__(self, inner_exception=None):
super(DBDeadlock, self).__init__(inner_exception)

37
oslo/db/sqlalchemy/exc_filters.py
View File

@ -161,6 +161,43 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
raise exception.DBDuplicateEntry(columns, integrity_error)
@filters("sqlite", sqla_exc.IntegrityError,
r".*SQL error: foreign key constraint failed")
@filters("postgresql", sqla_exc.IntegrityError,
r".*on table \"(?P<table>[^\"]+)\" violates "
"foreign key constraint \"(?P<constraint>[^\"]+)\"\s*\n"
"DETAIL: Key \((?P<key>.+)\)=\(.+\) "
"is not present in table "
"\"(?P<key_table>[^\"]+)\".")
@filters("mysql", sqla_exc.IntegrityError,
r".* Cannot add or update a child row: "
"a foreign key constraint fails "
"\((?P<table>.+), CONSTRAINT (?P<constraint>.+) "
"FOREIGN KEY \((?P<key>.+)\) "
"REFERENCES (?P<key_table>.+) \(.+\)\)")
def _foreign_key_error(integrity_error, match, engine_name, is_disconnect):
"""Filter for foreign key errors."""
try:
table = match.group("table")
except IndexError:
table = None
try:
constraint = match.group("constraint")
except IndexError:
constraint = None
try:
key = match.group("key")
except IndexError:
key = None
try:
key_table = match.group("key_table")
except IndexError:
key_table = None
raise exception.DBReferenceError(table, constraint, key, key_table,
integrity_error)
@filters("ibm_db_sa", sqla_exc.IntegrityError, r"^.*SQL0803N.*$")
def _db2_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
"""Filter for DB2 duplicate key errors.

83
tests/sqlalchemy/test_exc_filters.py
View File

@ -181,6 +181,89 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
self.assertEqual("mysqldb has an attribute error", matched.message)
class TestRaiseReferenceError(TestsExceptionFilter):
def test_postgresql(self):
e = self._run_test(
"postgresql",
"INSERT SOMETHING",
self.IntegrityError(
"insert or update on table "
"\"resource_entity\" "
"violates foreign key constraint "
"\"resource_entity_entity_id_fkey\"\n"
"DETAIL: Key "
"(entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
"is not present in table \"entity\".\n"
"'INSERT INTO resource_entity (resource_id, entity_id, name) "
"VALUES (%(resource_id)s, "
"%(entity_id)s, %(name)s)' "
"{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
"'name': u'foo', "
"'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'}"),
exception.DBReferenceError,
)
self.assertEqual("resource_entity", e.table)
self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
self.assertEqual("entity_id", e.key)
self.assertEqual("entity", e.key_table)
self.assertEqual(
"(IntegrityError) insert or update on table "
"\"resource_entity\" violates foreign key constraint "
"\"resource_entity_entity_id_fkey\"\n"
"DETAIL: Key (entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
"is not present in table \"entity\".\n"
"'INSERT INTO resource_entity (resource_id, entity_id, name) "
"VALUES (%(resource_id)s, %(entity_id)s, %(name)s)' "
"{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
"'name': u'foo', "
"'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'} "
"'INSERT SOMETHING' ()",
str(e))
def test_mysql(self):
e = self._run_test(
"mysql",
"INSERT SOMETHING",
self.IntegrityError(
"Cannot add or update a child row: "
"a foreign key constraint fails "
"(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
"FOREIGN KEY (entity_id) "
"REFERENCES entity (entity_id))"
),
exception.DBReferenceError,
)
self.assertEqual("resource_entity", e.table)
self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
self.assertEqual("entity_id", e.key)
self.assertEqual("entity", e.key_table)
self.assertEqual(
"(IntegrityError) Cannot add or update a child row: "
"a foreign key constraint fails "
"(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
"FOREIGN KEY (entity_id) REFERENCES entity (entity_id)) "
"'INSERT SOMETHING' ()",
str(e))
def test_sqlite(self):
e = self._run_test(
"sqlite",
"INSERT SOMETHING",
self.IntegrityError(
"SQL error: foreign key constraint failed"
),
exception.DBReferenceError,
)
self.assertIsNone(e.table)
self.assertIsNone(e.constraint)
self.assertIsNone(e.key)
self.assertIsNone(e.key_table)
self.assertEqual(
"(IntegrityError) SQL error: foreign key "
"constraint failed 'INSERT SOMETHING' ()",
str(e))
class TestDuplicate(TestsExceptionFilter):
def _run_dupe_constraint_test(self, dialect_name, message,