openstack/oslo.db
Code Issues Proposed changes

Added handler for mysql 8.0.19 duplicate key error update

Browse Source 
In mysql 8.0.19 , Duplicate key error information is extended to
include the table name of the key.Previously, duplicate key error
information included only the key value and key name.

Unit tests are provided for updated changes.

Change-Id: Ic78059b625e73cece355541cb4d89e641abc1103
Closes-Bug: #1896916
This commit is contained in:
Kamlesh Chauvhan 2021-05-19 10:41:16 +00:00
parent 3400d2df48
commit 2c2e6d48ac
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',