openstack/deb-python-pymysql
Code Issues Proposed changes

Support %% and precedence spaces in Cursor.executemany()

Browse Source 
fixes #449
This commit is contained in:
INADA Naoki
2016-05-10 01:12:36 +09:00
parent 2ce27683fc de8524d0af
commit 784c5e511b
3 changed files with 20 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
pymysql/cursors.py
View File

@@ -12,8 +12,11 @@ from . import err
#: Regular expression for :meth:`Cursor.executemany`. #: Regular expression for :meth:`Cursor.executemany`.
#: executemany only suports simple bulk insert. #: executemany only suports simple bulk insert.
#: You can use it to load large dataset. #: 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_INSERT_VALUES = re.compile(
re.IGNORECASE | re.DOTALL) 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): class Cursor(object):
@@ -170,7 +173,7 @@ class Cursor(object):
m = RE_INSERT_VALUES.match(query) m = RE_INSERT_VALUES.match(query)
if m: if m:
q_prefix = m.group(1) q_prefix = m.group(1) % ()
q_values = m.group(2).rstrip() q_values = m.group(2).rstrip()
q_postfix = m.group(3) or '' q_postfix = m.group(3) or ''
assert q_values[0] == '(' and q_values[-1] == ')' assert q_values[0] == '(' and q_values[-1] == ')'

13
pymysql/tests/test_cursor.py
View File

@@ -99,3 +99,16 @@ class CursorTest(base.PyMySQLTestCase):
data_dict = [{'data': i} for i in xrange(10)] data_dict = [{'data': i} for i in xrange(10)]
cursor.executemany("insert into test (data) values (%(data)s)", data_dict) 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') 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")

1
runtests.py
View File

@@ -10,6 +10,7 @@ if not (PYPY or JYTHON or IRONPYTHON):
@atexit.register @atexit.register
def report_uncollectable(): def report_uncollectable():
import gc
if not gc.garbage: if not gc.garbage:
print("No garbages!") print("No garbages!")
return return