Add exception filter for _sqlite_dupe_key_error

We can get 'PRIMARY KEY must be unique' error on some
platforms and versions of sqlite library while trying to
insert the row with the same primary key that already exist.
In this case oslo.db should raise DBDuplicateEntry error.
Add corresponding filter to _sqlite_dupe_key_error

Closes-Bug: #1386145

Change-Id: Ifafd6a8e0b613a31e596043071aef4d410a976f2
This commit is contained in:
Oleksii Chuprykov
2014-10-27 13:33:11 +02:00
parent 8bb12c02d4
commit 4b2058b8c1
2 changed files with 22 additions and 3 deletions

View File

@@ -147,7 +147,8 @@ def _default_dupe_key_error(integrity_error, match, engine_name,
@filters("sqlite", sqla_exc.IntegrityError,
(r"^.*columns?(?P<columns>[^)]+)(is|are)\s+not\s+unique$",
r"^.*UNIQUE\s+constraint\s+failed:\s+(?P<columns>.+)$"))
r"^.*UNIQUE\s+constraint\s+failed:\s+(?P<columns>.+)$",
r"^.*PRIMARY\s+KEY\s+must\s+be\s+unique.*$"))
def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
"""Filter for SQLite duplicate key error.
@@ -162,9 +163,21 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
1 column - (IntegrityError) UNIQUE constraint failed: tbl.k1
N columns - (IntegrityError) UNIQUE constraint failed: tbl.k1, tbl.k2
sqlite since 3.8.2:
(IntegrityError) PRIMARY KEY must be unique
"""
columns = match.group('columns')
columns = [c.split('.')[-1] for c in columns.strip().split(", ")]
columns = []
# NOTE(ochuprykov): We can get here by last filter in which there are no
# groups. Trying to access the substring that matched by
# the group will lead to IndexError. In this case just
# pass empty list to exception.DBDuplicateEntry
try:
columns = match.group('columns')
columns = [c.split('.')[-1] for c in columns.strip().split(", ")]
except IndexError:
pass
raise exception.DBDuplicateEntry(columns, integrity_error)

View File

@@ -334,6 +334,12 @@ class TestDuplicate(TestsExceptionFilter):
"sqlite",
'UNIQUE constraint failed: tbl.a, tbl.b')
def test_sqlite_dupe_primary_key(self):
self._run_dupe_constraint_test(
"sqlite",
"PRIMARY KEY must be unique 'insert into t values(10)'",
expected_columns=[])
def test_mysql_mysqldb(self):
self._run_dupe_constraint_test(
"mysql",