openstack/oslo.db
Code Issues Proposed changes

Merge "Added handler for mysql 8.0.19 duplicate key error update"

Browse Source
This commit is contained in:
Zuul 2021-06-08 13:19:42 +00:00 committed by Gerrit Code Review
commit 2eed65f788
2 changed files with 35 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
oslo_db/sqlalchemy/exc_filters.py
View File

@ -120,6 +120,12 @@ def _default_dupe_key_error(integrity_error, match, engine_name,
N columns - (IntegrityError) duplicate key value violates unique
constraint "name_of_our_constraint"
mysql since 8.0.19:
1 column - (IntegrityError) (1062, "Duplicate entry 'value_of_c1' for key
'table_name.c1'")
N columns - (IntegrityError) (1062, "Duplicate entry 'values joined
with -' for key 'table_name.name_of_our_constraint'")
mysql+mysqldb:
1 column - (IntegrityError) (1062, "Duplicate entry 'value_of_c1' for key
'c1'")
@ -145,6 +151,9 @@ def _default_dupe_key_error(integrity_error, match, engine_name,
if not columns.startswith(uniqbase):
if engine_name == "postgresql":
columns = [columns[columns.index("_") + 1:columns.rindex("_")]]
elif (engine_name == "mysql") and \
(uniqbase in str(columns.split("0")[:1])):
columns = columns.split("0")[1:]
else:
columns = [columns]
else:

26
oslo_db/tests/sqlalchemy/test_exc_filters.py
View File

@ -821,6 +821,14 @@ class TestDuplicate(TestsExceptionFilter):
expected_value='2'
)
def test_mysql_duplicate_entry_key_start_with_tablename(self):
self._run_dupe_constraint_test(
"mysql",
"1062 (23000): Duplicate entry '2' for key 'tbl.uniq_tbl0b'",
expected_columns=['b'],
expected_value='2'
)
def test_mysql_binary(self):
self._run_dupe_constraint_test(
"mysql",
@ -839,6 +847,24 @@ class TestDuplicate(TestsExceptionFilter):
expected_value="'\\\\x8A$\\\\x8D\\\\xA6\"s\\\\x8E!,"
)
def test_mysql_duplicate_entry_key_start_with_tablename_binary(self):
self._run_dupe_constraint_test(
"mysql",
"(1062, \'Duplicate entry "
"\\\'\\\\x8A$\\\\x8D\\\\xA6\"s\\\\x8E\\\' "
"for key \\\'tbl.uniq_tbl0c1\\\'\')",
expected_columns=['c1'],
expected_value="\\\\x8A$\\\\x8D\\\\xA6\"s\\\\x8E"
)
self._run_dupe_constraint_test(
"mysql",
"(1062, \'Duplicate entry "
"''\\\\x8A$\\\\x8D\\\\xA6\"s\\\\x8E!,' "
"for key 'tbl.uniq_tbl0c1'\')",
expected_columns=['c1'],
expected_value="'\\\\x8A$\\\\x8D\\\\xA6\"s\\\\x8E!,"
)
def test_postgresql_single(self):
self._run_dupe_constraint_test(
'postgresql',