Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pymysql/cursors.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import re
2import warnings
4from . import err
6#: Regular expression for :meth:`Cursor.executemany`.
7#: executemany only supports simple bulk insert.
8#: You can use it to load large dataset.
9RE_INSERT_VALUES = re.compile(
10 r"\s*((?:INSERT|REPLACE)\b.+\bVALUES?\s*)"
11 + r"(\(\s*(?:%s|%\([^)]+\)s)\s*(?:,\s*(?:%s|%\([^)]+\)s)\s*)*\))"
12 + r'(\s*(?:AS\s+(?:`[^`]+`|"[^"]+"|[0-9A-Za-z_$]+)\s*'
13 + r'(?:\(\s*(?:`[^`]+`|"[^"]+"|[0-9A-Za-z_$]+)\s*'
14 + r'(?:,\s*(?:`[^`]+`|"[^"]+"|[0-9A-Za-z_$]+)\s*)*\))?\s*)?'
15 + r"(?:ON DUPLICATE.*)?);?\s*\Z",
16 re.IGNORECASE | re.DOTALL,
17)
20def _backquote_escape(s):
21 return s.replace("`", "``")
24class Cursor:
25 """
26 This is the object used to interact with the database.
28 Do not create an instance of a Cursor yourself. Call
29 connections.Connection.cursor().
31 See `Cursor <https://www.python.org/dev/peps/pep-0249/#cursor-objects>`_ in
32 the specification.
33 """
35 #: Max statement size which :meth:`executemany` generates.
36 #:
37 #: Max size of allowed statement is max_allowed_packet - packet_header_size.
38 #: Default value of max_allowed_packet is 1048576.
39 max_stmt_length = 1024000
41 def __init__(self, connection):
42 self.connection = connection
43 self.warning_count = 0
44 self.description = None
45 self.rownumber = 0
46 self.rowcount = -1
47 self.arraysize = 1
48 self._executed = None
49 self._result = None
50 self._rows = None
52 def close(self):
53 """
54 Closing a cursor just exhausts all remaining data.
55 """
56 conn = self.connection
57 if conn is None:
58 return
59 try:
60 while self.nextset():
61 pass
62 finally:
63 self.connection = None
65 def __enter__(self):
66 return self
68 def __exit__(self, *exc_info):
69 del exc_info
70 self.close()
72 def _get_db(self):
73 if not self.connection:
74 raise err.ProgrammingError("Cursor closed")
75 return self.connection
77 def _check_executed(self):
78 if not self._executed:
79 raise err.ProgrammingError("execute() first")
81 def _conv_row(self, row):
82 return row
84 def setinputsizes(self, *args):
85 """Does nothing, required by DB API."""
87 def setoutputsizes(self, *args):
88 """Does nothing, required by DB API."""
90 def _nextset(self, unbuffered=False):
91 """Get the next query set."""
92 conn = self._get_db()
93 current_result = self._result
94 if current_result is None or current_result is not conn._result:
95 return None
96 if not current_result.has_next:
97 return None
98 self._result = None
99 self._clear_result()
100 conn.next_result(unbuffered=unbuffered)
101 self._do_get_result()
102 return True
104 def nextset(self):
105 return self._nextset(False)
107 def _mogrify(self, query, args) -> str:
108 """Return query after binding args."""
109 escape = self._get_db().escape
111 if isinstance(args, dict):
112 args = {key: escape(item) for key, item in args.items()}
113 elif isinstance(args, (list, tuple)):
114 args = tuple(map(escape, args))
115 else:
116 # Escaping a single argument is not part of the DB-API or mysqlclient.
117 # It will be removed in the next version.
118 warnings.warn(
119 "single argument is deprecated and will be removed in the next version.",
120 DeprecationWarning,
121 stacklevel=3,
122 )
123 args = escape(args)
124 try:
125 return query % args
126 except TypeError as m:
127 raise err.ProgrammingError(str(m))
129 def mogrify(self, query, args=None) -> str:
130 """
131 Returns the exact string that would be sent to the database by calling the
132 execute() method.
134 :param query: Query to mogrify.
135 :type query: str
137 :param args: Parameters used with query. (optional)
138 :type args: tuple, list or dict
140 :return: The query with argument binding applied.
141 :rtype: str
143 This method follows the extension to the DB API 2.0 followed by Psycopg.
144 """
145 if args is None:
146 return query
148 return self._mogrify(query, args)
150 def execute(self, query, args=None):
151 """Execute a query.
153 :param query: Query to execute.
154 :type query: str
156 :param args: Parameters used with query. (optional)
157 :type args: tuple, list or dict
159 :return: Number of affected rows.
160 :rtype: int
162 If args is a list or tuple, %s can be used as a placeholder in the query.
163 If args is a dict, %(name)s can be used as a placeholder in the query.
164 """
165 while self.nextset():
166 pass
168 if args is not None:
169 query = self._mogrify(query, args)
170 result = self._query(query)
171 self._executed = query
172 return result
174 def executemany(self, query, args):
175 """Run several data against one query.
177 :param query: Query to execute.
178 :type query: str
180 :param args: Sequence of sequences or mappings. It is used as parameter.
181 :type args: tuple or list
183 :return: Number of rows affected, if any.
184 :rtype: int or None
186 This method improves performance on multiple-row INSERT and
187 REPLACE. Otherwise it is equivalent to looping over args with
188 execute().
189 """
190 if not args:
191 return
193 m = RE_INSERT_VALUES.match(query)
194 if m:
195 q_prefix = m.group(1) % ()
196 q_values = m.group(2).rstrip()
197 q_postfix = (m.group(3) or "") % ()
198 assert q_values[0] == "(" and q_values[-1] == ")"
199 return self._do_execute_many(
200 q_prefix,
201 q_values,
202 q_postfix,
203 args,
204 self.max_stmt_length,
205 self._get_db().encoding,
206 )
208 self.rowcount = sum(self.execute(query, arg) for arg in args)
209 return self.rowcount
211 def _do_execute_many(
212 self, prefix, values, postfix, args, max_stmt_length, encoding
213 ):
214 encoding = self._get_db().encoding
215 if isinstance(prefix, str):
216 prefix = prefix.encode(encoding)
217 if isinstance(postfix, str):
218 postfix = postfix.encode(encoding)
219 sql = bytearray(prefix)
220 args = iter(args)
221 sql += self._mogrify(values, next(args)).encode(encoding)
222 rows = 0
223 for arg in args:
224 v = self._mogrify(values, arg).encode(encoding)
225 if len(sql) + len(v) + len(postfix) + 1 > max_stmt_length:
226 rows += self.execute(sql + postfix)
227 sql = bytearray(prefix)
228 else:
229 sql += b","
230 sql += v
231 rows += self.execute(sql + postfix)
232 self.rowcount = rows
233 return rows
235 def callproc(self, procname, args=()):
236 """Execute stored procedure procname with args.
238 :param procname: Name of procedure to execute on server.
239 :type procname: str
241 :param args: Sequence of parameters to use with procedure.
242 :type args: tuple or list
244 Returns the original args.
246 Compatibility warning: PEP-249 specifies that any modified
247 parameters must be returned. This is currently impossible
248 as they are only available by storing them in a server
249 variable and then retrieved by a query. Since stored
250 procedures return zero or more result sets, there is no
251 reliable way to get at OUT or INOUT parameters via callproc.
252 The server variables are named @_procname_n, where procname
253 is the parameter above and n is the position of the parameter
254 (from zero). Once all result sets generated by the procedure
255 have been fetched, you can issue a SELECT @_procname_0, ...
256 query using .execute() to get any OUT or INOUT values.
258 Compatibility warning: The act of calling a stored procedure
259 itself creates an empty result set. This appears after any
260 result sets generated by the procedure. This is non-standard
261 behavior with respect to the DB-API. Be sure to use nextset()
262 to advance through all result sets; otherwise you may get
263 disconnected.
264 """
265 procname_escaped = _backquote_escape(procname)
266 conn = self._get_db()
268 if args:
269 fmt = f"@`_{procname_escaped}_%d`=%s"
270 self._query(
271 "SET %s"
272 % ",".join(
273 fmt % (index, conn.escape(arg)) for index, arg in enumerate(args)
274 )
275 )
276 self.nextset()
278 q = "CALL `{}`({})".format(
279 procname_escaped,
280 ",".join([f"@`_{procname_escaped}_{i}`" for i in range(len(args))]),
281 )
282 self._query(q)
283 self._executed = q
284 return args
286 def fetchone(self):
287 """Fetch the next row."""
288 self._check_executed()
289 if self._rows is None or self.rownumber >= len(self._rows):
290 return None
291 result = self._rows[self.rownumber]
292 self.rownumber += 1
293 return result
295 def fetchmany(self, size=None):
296 """Fetch several rows."""
297 self._check_executed()
298 if self._rows is None:
299 # Django expects () for EOF.
300 # https://github.com/django/django/blob/0c1518ee429b01c145cf5b34eab01b0b92f8c246/django/db/backends/mysql/features.py#L8
301 return ()
302 end = self.rownumber + (size or self.arraysize)
303 result = self._rows[self.rownumber : end]
304 self.rownumber = min(end, len(self._rows))
305 return result
307 def fetchall(self):
308 """Fetch all the rows."""
309 self._check_executed()
310 if self._rows is None:
311 return []
312 if self.rownumber:
313 result = self._rows[self.rownumber :]
314 else:
315 result = self._rows
316 self.rownumber = len(self._rows)
317 return result
319 def scroll(self, value, mode="relative"):
320 self._check_executed()
321 if mode == "relative":
322 r = self.rownumber + value
323 elif mode == "absolute":
324 r = value
325 else:
326 raise err.ProgrammingError("unknown scroll mode %s" % mode)
328 if not (0 <= r < len(self._rows)):
329 raise IndexError("out of range")
330 self.rownumber = r
332 def _query(self, q):
333 conn = self._get_db()
334 self._clear_result()
335 conn.query(q)
336 self._do_get_result()
337 return self.rowcount
339 def _clear_result(self):
340 self.rownumber = 0
341 self._result = None
343 self.rowcount = 0
344 self.warning_count = 0
345 self.description = None
346 self.lastrowid = None
347 self._rows = None
349 def _do_get_result(self):
350 conn = self._get_db()
352 self._result = result = conn._result
354 self.rowcount = result.affected_rows
355 self.warning_count = result.warning_count
356 self.description = result.description
357 self.lastrowid = result.insert_id
358 self._rows = result.rows
360 def __iter__(self):
361 return self
363 def __next__(self):
364 row = self.fetchone()
365 if row is None:
366 raise StopIteration
367 return row
370class DictCursorMixin:
371 # You can override this to use OrderedDict or other dict-like types.
372 dict_type = dict
374 def _do_get_result(self):
375 super()._do_get_result()
376 fields = []
377 if self.description:
378 for f in self._result.fields:
379 name = f.name
380 if name in fields:
381 name = f.table_name + "." + name
382 fields.append(name)
383 self._fields = fields
385 if fields and self._rows:
386 self._rows = [self._conv_row(r) for r in self._rows]
388 def _conv_row(self, row):
389 if row is None:
390 return None
391 return self.dict_type(zip(self._fields, row))
394class DictCursor(DictCursorMixin, Cursor):
395 """A cursor which returns results as a dictionary"""
398class SSCursor(Cursor):
399 """
400 Unbuffered Cursor, mainly useful for queries that return a lot of data,
401 or for connections to remote servers over a slow network.
403 Instead of copying every row of data into a buffer, this will fetch
404 rows as needed. The upside of this is the client uses much less memory,
405 and rows are returned much faster when traveling over a slow network
406 or if the result set is very big.
408 There are limitations, though. The MySQL protocol doesn't support
409 returning the total number of rows, so the only way to tell how many rows
410 there are is to iterate over every row returned. Also, it currently isn't
411 possible to scroll backwards, as only the current row is held in memory.
412 """
414 def _conv_row(self, row):
415 return row
417 def close(self):
418 conn = self.connection
419 if conn is None:
420 return
422 if self._result is not None and self._result is conn._result:
423 self._result._finish_unbuffered_query()
425 try:
426 while self.nextset():
427 pass
428 finally:
429 self.connection = None
431 __del__ = close
433 def _query(self, q):
434 conn = self._get_db()
435 self._clear_result()
436 conn.query(q, unbuffered=True)
437 self._do_get_result()
438 return self.rowcount
440 def nextset(self):
441 return self._nextset(unbuffered=True)
443 def read_next(self):
444 """Read next row."""
445 return self._conv_row(self._result._read_rowdata_packet_unbuffered())
447 def fetchone(self):
448 """Fetch next row."""
449 self._check_executed()
450 row = self.read_next()
451 if row is None:
452 self.warning_count = self._result.warning_count
453 return None
454 self.rownumber += 1
455 return row
457 def fetchall(self):
458 """
459 Fetch all, as per MySQLdb. Pretty useless for large queries, as
460 it is buffered. See fetchall_unbuffered(), if you want an unbuffered
461 generator version of this method.
462 """
463 return list(self.fetchall_unbuffered())
465 def fetchall_unbuffered(self):
466 """
467 Fetch all, implemented as a generator, which isn't to standard,
468 however, it doesn't make sense to return everything in a list, as that
469 would use ridiculous memory for large result sets.
470 """
471 return iter(self.fetchone, None)
473 def fetchmany(self, size=None):
474 """Fetch many."""
475 self._check_executed()
476 if size is None:
477 size = self.arraysize
479 rows = []
480 for i in range(size):
481 row = self.read_next()
482 if row is None:
483 self.warning_count = self._result.warning_count
484 break
485 rows.append(row)
486 self.rownumber += 1
487 if not rows:
488 # Django expects () for EOF.
489 # https://github.com/django/django/blob/0c1518ee429b01c145cf5b34eab01b0b92f8c246/django/db/backends/mysql/features.py#L8
490 return ()
491 return rows
493 def scroll(self, value, mode="relative"):
494 self._check_executed()
496 if mode == "relative":
497 if value < 0:
498 raise err.NotSupportedError(
499 "Backwards scrolling not supported by this cursor"
500 )
502 for _ in range(value):
503 self.read_next()
504 self.rownumber += value
505 elif mode == "absolute":
506 if value < self.rownumber:
507 raise err.NotSupportedError(
508 "Backwards scrolling not supported by this cursor"
509 )
511 end = value - self.rownumber
512 for _ in range(end):
513 self.read_next()
514 self.rownumber = value
515 else:
516 raise err.ProgrammingError("unknown scroll mode %s" % mode)
519class SSDictCursor(DictCursorMixin, SSCursor):
520 """An unbuffered cursor, which returns results as a dictionary"""