diff --git a/pymysql/cursors.py b/pymysql/cursors.py index ff00d7b..18b6a25 100644 --- a/pymysql/cursors.py +++ b/pymysql/cursors.py @@ -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) diff --git a/pymysql/tests/test_cursor.py b/pymysql/tests/test_cursor.py index f900774..34bcaa6 100644 --- a/pymysql/tests/test_cursor.py +++ b/pymysql/tests/test_cursor.py @@ -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')