diff options
| -rw-r--r-- | oslo/db/exception.py | 11 | ||||
| -rw-r--r-- | oslo/db/sqlalchemy/exc_filters.py | 37 | ||||
| -rw-r--r-- | tests/sqlalchemy/test_exc_filters.py | 83 |
3 files changed, 131 insertions, 0 deletions
diff --git a/oslo/db/exception.py b/oslo/db/exception.py index 4adf350..4188481 100644 --- a/oslo/db/exception.py +++ b/oslo/db/exception.py @@ -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) diff --git a/oslo/db/sqlalchemy/exc_filters.py b/oslo/db/sqlalchemy/exc_filters.py index 3ec19cc..cdc8827 100644 --- a/oslo/db/sqlalchemy/exc_filters.py +++ b/oslo/db/sqlalchemy/exc_filters.py @@ -156,6 +156,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. diff --git a/tests/sqlalchemy/test_exc_filters.py b/tests/sqlalchemy/test_exc_filters.py index 649716c..25908d5 100644 --- a/tests/sqlalchemy/test_exc_filters.py +++ b/tests/sqlalchemy/test_exc_filters.py @@ -166,6 +166,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, |