diff --git a/pymysql/cursors.py b/pymysql/cursors.py index 69a41f4..f17611b 100644 --- a/pymysql/cursors.py +++ b/pymysql/cursors.py @@ -12,8 +12,11 @@ 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)\s*)*\))(\s*(?:ON DUPLICATE.*)?)\Z""", - re.IGNORECASE | re.DOTALL) +RE_INSERT_VALUES = re.compile( + r"\s*(INSERT\s.+\sVALUES\s+)" + + r"(\(\s*(?:%s|%\(.+\)s)\s*(?:,\s*(?:%s|%\(.+\)s)\s*)*\))" + + r"(\s*(?:ON DUPLICATE.*)?)\Z", + re.IGNORECASE | re.DOTALL) class Cursor(object): @@ -170,7 +173,7 @@ class Cursor(object): m = RE_INSERT_VALUES.match(query) if m: - q_prefix = m.group(1) + q_prefix = m.group(1) % () q_values = m.group(2).rstrip() q_postfix = m.group(3) or '' assert q_values[0] == '(' and q_values[-1] == ')' diff --git a/pymysql/tests/test_cursor.py b/pymysql/tests/test_cursor.py index 34bcaa6..fb0178b 100644 --- a/pymysql/tests/test_cursor.py +++ b/pymysql/tests/test_cursor.py @@ -99,3 +99,16 @@ class CursorTest(base.PyMySQLTestCase): 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') + + # %% in column set + cursor.execute("""\ + CREATE TABLE percent_test ( + `A%` INTEGER, + `B%` INTEGER)""") + try: + q = "INSERT INTO percent_test (`A%%`, `B%%`) VALUES (%s, %s)" + self.assertIsNotNone(pymysql.cursors.RE_INSERT_VALUES.match(q) + cursor.executemany(q, [(3, 4), (5, 6)]) + self.assertTrue(cursor._executed.endswith("(3, 4),(5, 6)"), "executemany with %% not in one query") + finally: + cursor.execute("DROP TABLE IF EXISTS percent_test") diff --git a/runtests.py b/runtests.py index 367e6ac..00e492b 100755 --- a/runtests.py +++ b/runtests.py @@ -10,6 +10,7 @@ if not (PYPY or JYTHON or IRONPYTHON): @atexit.register def report_uncollectable(): + import gc if not gc.garbage: print("No garbages!") return