openstack/deb-python-oslo.db
Code Issues Proposed changes

Merge "Add DBDuplicateEntry detection for mysqlconnector driver"

Browse Source
This commit is contained in:
Jenkins 2014-07-14 22:00:01 +00:00 committed by Gerrit Code Review
commit 44bd28704c
2 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
oslo/db/sqlalchemy/session.py
View File

@ -328,12 +328,20 @@ class SqliteForeignKeysListener(PoolListener):
# N columns - (IntegrityError) duplicate key value violates unique
# constraint "name_of_our_constraint"
#
# mysql:
# mysql+mysqldb:
# 1 column - (IntegrityError) (1062, "Duplicate entry 'value_of_c1' for key
# 'c1'")
# N columns - (IntegrityError) (1062, "Duplicate entry 'values joined
# with -' for key 'name_of_our_constraint'")
#
# mysql+mysqlconnector:
# http://docs.sqlalchemy.org/en/rel_0_9/dialects/
# mysql.html#module-sqlalchemy.dialects.mysql.mysqlconnector
# 1 column - (IntegrityError) 1062 (23000): Duplicate entry 'value_of_c1' for
# key 'c1'
# N columns - (IntegrityError) 1062 (23000): Duplicate entry 'values
# joined with -' for key 'name_of_our_constraint'
#
# ibm_db_sa:
# N columns - (IntegrityError) SQL0803N One or more values in the INSERT
# statement, UPDATE statement, or foreign key update caused by a
@ -344,8 +352,9 @@ class SqliteForeignKeysListener(PoolListener):
_DUP_KEY_RE_DB = {
"sqlite": (re.compile(r"^.*columns?([^)]+)(is|are)\s+not\s+unique$"),
re.compile(r"^.*UNIQUE\s+constraint\s+failed:\s+(.+)$")),
"postgresql": (re.compile(r"^.*duplicate\s+key.*\"([^\"]+)\"\s*\n.*$"),),
"mysql": (re.compile(r"^.*\(1062,.*'([^\']+)'\"\)$"),),
"postgresql": (re.compile(r'^.*duplicate\s+key.*"([^"]+)"\s*\n.*$'),),
"mysql": (re.compile(
r"^.*\b1062\b.*Duplicate entry '[^']+' for key '([^']+)'.*$"),),
"ibm_db_sa": (re.compile(r"^.*SQL0803N.*$"),),
}

9
tests/sqlalchemy/test_utils.py
View File

@ -719,13 +719,20 @@ class TestRaiseDuplicateEntryError(test_base.BaseTestCase):
'(IntegrityError) UNIQUE constraint failed: tbl.a, tbl.b'
)
def test_mysql(self):
def test_mysql_mysqldb(self):
self._test_impl(
'mysql',
'(IntegrityError) (1062, "Duplicate entry '
'\'2-3\' for key \'uniq_tbl0a0b\'")'
)
def test_mysql_mysqlconnector(self):
self._test_impl(
'mysql',
'(IntegrityError) 1062 (23000): Duplicate entry '
'\'2-3\' for key \'uniq_tbl0a0b\'',
)
def test_postgresql(self):
self._test_impl(
'postgresql',