openstack/deb-python-pymysql
Code Issues Proposed changes

Merge pull request #427 from WorldException/master

Browse Source 
Add %(name)s paramstyle support for executemany
This commit is contained in:
INADA Naoki
2016-02-24 19:11:35 +09:00
parent 2622e5277f 07a7be71c9
commit 9724752d3c
2 changed files with 31 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pymysql/cursors.py
View File

@@ -12,7 +12,7 @@ from . import err
#: Regular expression for :meth:`Cursor.executemany`.
#: executemany only suports simple bulk insert.
#: You can use it to load large dataset.
RE_INSERT_VALUES = re.compile(r"""(INSERT\s.+\sVALUES\s+)(\(\s*%s\s*(?:,\s*%s\s*)*\))(\s*(?:ON DUPLICATE.*)?)\Z""",
RE_INSERT_VALUES = re.compile(r"""(INSERT\s.+\sVALUES\s+)(\(\s*(?:%s|%\(.+\)s)\s*(?:,\s*(?:%s|%\(.+\)s)\s*)*\))(\s*(?:ON DUPLICATE.*)?)\Z""",
re.IGNORECASE | re.DOTALL)

30
pymysql/tests/test_cursor.py
View File

@@ -69,3 +69,33 @@ class CursorTest(base.PyMySQLTestCase):
)
self.assertIsNone(c2.fetchone())
def test_executemany(self):
conn = self.test_connection
cursor = conn.cursor(pymysql.cursors.Cursor)
m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%s, %s)")
self.assertIsNotNone(m, 'error parse %s')
self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')
m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id)s, %(name)s)")
self.assertIsNotNone(m, 'error parse %(name)s')
self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')
m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id_name)s, %(name)s)")
self.assertIsNotNone(m, 'error parse %(id_name)s')
self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')
m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id_name)s, %(name)s) ON duplicate update")
self.assertIsNotNone(m, 'error parse %(id_name)s')
self.assertEqual(m.group(3), ' ON duplicate update', 'group 3 not ON duplicate update, bug in RE_INSERT_VALUES?')
# cursor._executed myst bee "insert into test (data) values (0),(1),(2),(3),(4),(5),(6),(7),(8),(9)"
# list args
data = xrange(10)
cursor.executemany("insert into test (data) values (%s)", data)
self.assertTrue(cursor._executed.endswith(",(7),(8),(9)"), 'execute many with %s not in one query')
# dict args
data_dict = [{'data': i} for i in xrange(10)]
cursor.executemany("insert into test (data) values (%(data)s)", data_dict)
self.assertTrue(cursor._executed.endswith(",(7),(8),(9)"), 'execute many with %(data)s not in one query')