"""MySQLdb Cursors This module implements Cursors of various types for MySQLdb. By default, MySQLdb uses the Cursor class. """ import re from ._exceptions import InternalError, MySQLError, ProgrammingError from .constants import CLIENT, CR _EXECUTEMANY_MULTI_SEPARATOR = b"\n;\n" #: Regular expression for ``Cursor.executemany```. #: executemany only supports simple bulk insert. #: You can use it to load large dataset. RE_INSERT_VALUES = re.compile( "".join( [ r"\s*((?:INSERT|REPLACE)\b.+\bVALUES?\s*)", r"(\(\s*(?:%s|%\(.+\)s)\s*(?:,\s*(?:%s|%\(.+\)s)\s*)*\))", r"(\s*(?:ON DUPLICATE.*)?);?\s*\Z", ] ), re.IGNORECASE | re.DOTALL, ) _RE_INSERT_VALUES_BYTES = re.compile( RE_INSERT_VALUES.pattern.encode("ascii"), re.IGNORECASE | re.DOTALL ) _RE_EXECUTEMANY_DML = re.compile( r"\s*(?:INSERT|REPLACE|UPDATE|DELETE)\b", re.IGNORECASE ) _RE_EXECUTEMANY_DML_BYTES = re.compile( _RE_EXECUTEMANY_DML.pattern.encode("ascii"), re.IGNORECASE ) _RE_RETURNING = re.compile(r"\bRETURNING\b", re.IGNORECASE) _RE_RETURNING_BYTES = re.compile( _RE_RETURNING.pattern.encode("ascii"), re.IGNORECASE ) def _match_insert_values(query): if isinstance(query, (bytes, bytearray)): return _RE_INSERT_VALUES_BYTES.match(query) return RE_INSERT_VALUES.match(query) def _is_executemany_dml(query): """Return whether query is safe for client-side multi-statement batching.""" if isinstance(query, (bytes, bytearray)): return ( b";" not in query and _RE_EXECUTEMANY_DML_BYTES.match(query) is not None and _RE_RETURNING_BYTES.search(query) is None ) return ( ";" not in query and _RE_EXECUTEMANY_DML.match(query) is not None and _RE_RETURNING.search(query) is None ) def _backquote_escape(s): return s.replace(b"`", b"``") class BaseCursor: """A base for Cursor classes. Useful attributes: description A tuple of DB API 7-tuples describing the columns in the last executed query; see PEP-249 for details. description_flags Tuple of column flags for last query, one entry per column in the result set. Values correspond to those in MySQLdb.constants.FLAG. See MySQL documentation (C API) for more information. Non-standard extension. arraysize default number of rows fetchmany() will fetch """ #: Max statement size which :meth:`executemany` generates. #: #: Max size of allowed statement is max_allowed_packet - packet_header_size. #: Default value of max_allowed_packet is 1048576. max_stmt_length = 64 * 1024 #: Maximum encoded size and statement count for multi-statement #: ``executemany`` fallback batches. The size includes separators and is #: measured after argument conversion. Subclasses may override them. max_multi_stmt_length = 16_000 max_multi_stmt_count = 200 #: Override with ``"loop"`` or ``"multi"`` on a cursor subclass or #: instance. ``None`` inherits the policy from the connection. executemany_fallback = None connection = None def __init__(self, connection): self.connection = connection self.warning_count = 0 self.description = None self.description_flags = None self.rowcount = 0 self.arraysize = 1 self._executed = None self.lastrowid = None self._result = None self.rownumber = None self._rows = None def _discard(self): self.description = None self.description_flags = None # Django uses some member after __exit__. # So we keep rowcount and lastrowid here. They are cleared in Cursor._query(). # self.rowcount = 0 # self.lastrowid = None self._rows = None self.rownumber = None if self._result: self._result.discard() self._result = None con = self.connection if con is None: return while con.next_result() == 0: # -1 means no more data. con.discard_result() def close(self): """Close the cursor. No further queries will be possible.""" try: if self.connection is None: return self._discard() finally: self.connection = None self._result = None def __enter__(self): return self def __exit__(self, *exc_info): del exc_info self.close() def _check_executed(self): if not self._executed: raise ProgrammingError("execute() first") def nextset(self): """Advance to the next result set. Returns None if there are no more result sets. """ if self._executed: self.fetchall() db = self._get_db() nr = db.next_result() if nr == -1: return None self._do_get_result(db) self._post_get_result() return 1 def _do_get_result(self, db): self._result = result = self._get_result() if result is None: self.description = self.description_flags = None else: self.description = result.describe() self.description_flags = result.field_flags() self.warning_count = db.warning_count() self.rowcount = db.affected_rows() self.rownumber = 0 self.lastrowid = db.insert_id() def _post_get_result(self): pass def setinputsizes(self, *args): """Does nothing, required by DB API.""" def setoutputsizes(self, *args): """Does nothing, required by DB API.""" def _get_db(self): con = self.connection if con is None: raise ProgrammingError("cursor closed") return con def execute(self, query, args=None): """Execute a query. query -- string, query to execute on server args -- optional sequence or mapping, parameters to use with query. Note: If args is a sequence, then %s must be used as the parameter placeholder in the query. If a mapping is used, %(key)s must be used as the placeholder. Returns integer represents rows affected, if any """ self._discard() mogrified_query = self._mogrify(query, args) assert isinstance(mogrified_query, (bytes, bytearray)) res = self._query(mogrified_query) return res def _mogrify(self, query, args=None): """Return query after binding args.""" db = self._get_db() if isinstance(query, str): query = query.encode(db.encoding) if args is not None: if isinstance(args, dict): nargs = {} for key, item in args.items(): if isinstance(key, str): key = key.encode(db.encoding) nargs[key] = db.literal(item) args = nargs else: args = tuple(map(db.literal, args)) try: query = query % args except TypeError as m: raise ProgrammingError(str(m)) return query def mogrify(self, query, args=None): """Return query after binding args. query -- string, query to mogrify args -- optional sequence or mapping, parameters to use with query. Note: If args is a sequence, then %s must be used as the parameter placeholder in the query. If a mapping is used, %(key)s must be used as the placeholder. Returns string representing query that would be executed by the server """ return self._mogrify(query, args).decode(self._get_db().encoding) def executemany(self, query, args): # type: (str, list) -> int """Execute a multi-row query. :param query: query to execute on server :param args: Sequence of sequences or mappings. It is used as parameter. :return: Number of rows affected, if any. This method improves performance on multiple-row INSERT and REPLACE. When ``executemany_fallback`` is ``"multi"``, it also batches safe DML statements if the connection has multi statements enabled. Otherwise, it is equivalent to looping over args with execute(). """ if not args: return m = _match_insert_values(query) if m: q_prefix = m.group(1) % () q_values = m.group(2).rstrip() q_postfix = m.group(3) or "" assert q_values[:1] in ("(", b"(") assert q_values[-1:] in (")", b")") return self._do_execute_many( q_prefix, q_values, q_postfix, args, self.max_stmt_length, self._get_db().encoding, ) fallback = self.executemany_fallback db = self._get_db() if fallback is None: fallback = getattr(db, "executemany_fallback", "loop") if fallback not in ("loop", "multi"): raise ValueError("executemany_fallback must be either 'loop' or 'multi'") if ( fallback == "multi" and db.client_flag & CLIENT.MULTI_STATEMENTS and _is_executemany_dml(query) ): return self._do_execute_many_multi(query, args) rows = None for arg in args: result = self.execute(query, arg) rows = result if rows is None else rows + result if rows is None: return self.rowcount = rows return self.rowcount def _do_execute_many_multi(self, query, args): rows = 0 statement_count = 0 sql = bytearray() for arg in args: statement = self._mogrify(query, arg) if statement_count and ( statement_count >= self.max_multi_stmt_count or len(sql) + len(_EXECUTEMANY_MULTI_SEPARATOR) + len(statement) > self.max_multi_stmt_length ): rows += self._execute_multi_statement_batch( bytes(sql), statement_count ) sql.clear() statement_count = 0 if statement_count: sql += _EXECUTEMANY_MULTI_SEPARATOR sql += statement statement_count += 1 if not statement_count: return rows += self._execute_multi_statement_batch(bytes(sql), statement_count) self.rowcount = rows return rows def _execute_multi_statement_batch(self, query, statement_count): """Execute and fully consume one generated multi-statement query.""" if statement_count == 1: return self.execute(query) db = self._get_db() query_started = False try: query_started = True self.execute(query) if self.description is not None: self._raise_multi_statement_result_mismatch(db) rows = self.rowcount for _ in range(statement_count - 1): if not db.more_results(): self._raise_multi_statement_result_mismatch(db) if db.next_result() != 0: self._raise_multi_statement_result_mismatch(db) self._do_get_result(db) if self.description is not None: self._raise_multi_statement_result_mismatch(db) self._post_get_result() rows += self.rowcount if db.more_results(): self._raise_multi_statement_result_mismatch(db) return rows except BaseException as exc: # A server-side SQL error from next_result() terminates the rest of # the multi-statement query and leaves the protocol synchronized. # Interruptions and client/protocol failures can leave unread # results, so discard the connection instead of risking reuse. self.description = None self.description_flags = None self.warning_count = 0 self.rowcount = None self.lastrowid = None self._result = None self._rows = None self.rownumber = None if query_started and self._multi_statement_error_needs_close(exc): self._close_connection(db) raise def _raise_multi_statement_result_mismatch(self, db): if self._result is not None: try: self._result.discard() except BaseException: # noqa: S110 pass self._result = None self._close_connection(db) raise InternalError("multi-statement executemany result count mismatch") @staticmethod def _close_connection(db): try: db.close() except BaseException: # noqa: S110 pass def _multi_statement_error_needs_close(self, exc): if not isinstance(exc, MySQLError): return True if not exc.args or not isinstance(exc.args[0], int): return True errno = exc.args[0] return ( CR.MIN_ERROR <= errno <= CR.MAX_ERROR or errno == 1153 # ER_NET_PACKET_TOO_LARGE or errno == 1927 # ER_CONNECTION_KILLED (MariaDB) or errno == 4031 # ER_CLIENT_INTERACTION_TIMEOUT ) def _do_execute_many( self, prefix, values, postfix, args, max_stmt_length, encoding ): if isinstance(prefix, str): prefix = prefix.encode(encoding) if isinstance(values, str): values = values.encode(encoding) if isinstance(postfix, str): postfix = postfix.encode(encoding) sql = bytearray(prefix) args = iter(args) try: first_arg = next(args) except StopIteration: return v = self._mogrify(values, first_arg) sql += v rows = 0 for arg in args: v = self._mogrify(values, arg) if len(sql) + len(v) + len(postfix) + 1 > max_stmt_length: rows += self.execute(sql + postfix) sql = bytearray(prefix) else: sql += b"," sql += v rows += self.execute(sql + postfix) self.rowcount = rows return rows def callproc(self, procname, args=()): """Execute stored procedure procname with args procname -- string, name of procedure to execute on server args -- Sequence of parameters to use with procedure Returns the original args. Compatibility warning: PEP-249 specifies that any modified parameters must be returned. This is currently impossible as they are only available by storing them in a server variable and then retrieved by a query. Since stored procedures return zero or more result sets, there is no reliable way to get at OUT or INOUT parameters via callproc. The server variables are named @`_procname_n`, where procname is the parameter above and n is the position of the parameter (from zero). Once all result sets generated by the procedure have been fetched, you can issue a SELECT @`_procname_0`, ... query using .execute() to get any OUT or INOUT values. Compatibility warning: The act of calling a stored procedure itself creates an empty result set. This appears after any result sets generated by the procedure. This is non-standard behavior with respect to the DB-API. Be sure to use nextset() to advance through all result sets; otherwise you may get disconnected. """ db = self._get_db() if isinstance(procname, str): procname = procname.encode(db.encoding) procname_escaped = _backquote_escape(procname) if args: fmt = b"@`_" + procname_escaped + b"_%d`=%s" q = b"SET %s" % b",".join( fmt % (index, db.literal(arg)) for index, arg in enumerate(args) ) self._query(q) self.nextset() q = b"CALL `%s`(%s)" % ( procname_escaped, b",".join([b"@`_%s_%d`" % (procname_escaped, i) for i in range(len(args))]), ) self._query(q) return args def _query(self, q): db = self._get_db() self._result = None self.warning_count = 0 self.rowcount = None self.lastrowid = None db.query(q) self._do_get_result(db) self._post_get_result() self._executed = q return self.rowcount def _fetch_row(self, size=1): if not self._result: return () return self._result.fetch_row(size, self._fetch_type) def __iter__(self): return self def __next__(self): row = self.fetchone() if row is None: raise StopIteration return row def __getattr__(self, name): # DB-API 2.0 optional extension says these errors can be accessed # via Connection object. But MySQLdb had defined them on Cursor object. import warnings from . import _exceptions as err if name in ( "Warning", "Error", "InterfaceError", "DatabaseError", "DataError", "OperationalError", "IntegrityError", "InternalError", "ProgrammingError", "NotSupportedError", ): # Deprecated since v2.3. Will be removed in 2027. warnings.warn( "errors should be accessed from `MySQLdb` package", DeprecationWarning, stacklevel=2, ) return getattr(err, name) raise AttributeError(name) class CursorStoreResultMixIn: """This is a MixIn class which causes the entire result set to be stored on the client side, i.e. it uses mysql_store_result(). If the result set can be very large, consider adding a LIMIT clause to your query, or using CursorUseResultMixIn instead.""" def _get_result(self): return self._get_db().store_result() def _post_get_result(self): self._rows = self._fetch_row(0) self._result = None def fetchone(self): """Fetches a single row from the cursor. None indicates that no more rows are available.""" self._check_executed() if self.rownumber >= len(self._rows): return None result = self._rows[self.rownumber] self.rownumber = self.rownumber + 1 return result def fetchmany(self, size=None): """Fetch up to size rows from the cursor. Result set may be smaller than size. If size is not defined, cursor.arraysize is used.""" self._check_executed() end = self.rownumber + (size or self.arraysize) result = self._rows[self.rownumber : end] self.rownumber = min(end, len(self._rows)) return result def fetchall(self): """Fetches all available rows from the cursor.""" self._check_executed() if self.rownumber: result = self._rows[self.rownumber :] else: result = self._rows self.rownumber = len(self._rows) return result def scroll(self, value, mode="relative"): """Scroll the cursor in the result set to a new position according to mode. If mode is 'relative' (default), value is taken as offset to the current position in the result set, if set to 'absolute', value states an absolute target position.""" self._check_executed() if mode == "relative": r = self.rownumber + value elif mode == "absolute": r = value else: raise ProgrammingError("unknown scroll mode %s" % repr(mode)) if r < 0 or r >= len(self._rows): raise IndexError("out of range") self.rownumber = r class CursorUseResultMixIn: """This is a MixIn class which causes the result set to be stored in the server and sent row-by-row to client side, i.e. it uses mysql_use_result(). You MUST retrieve the entire result set and close() the cursor before additional queries can be performed on the connection.""" def _get_result(self): return self._get_db().use_result() def fetchone(self): """Fetches a single row from the cursor.""" self._check_executed() r = self._fetch_row(1) if not r: self.warning_count = self._get_db().warning_count() return None self.rownumber = self.rownumber + 1 return r[0] def fetchmany(self, size=None): """Fetch up to size rows from the cursor. Result set may be smaller than size. If size is not defined, cursor.arraysize is used.""" self._check_executed() size = size or self.arraysize r = self._fetch_row(size) if len(r) < size: self.warning_count = self._get_db().warning_count() self.rownumber = self.rownumber + len(r) return r def fetchall(self): """Fetches all available rows from the cursor.""" self._check_executed() r = self._fetch_row(0) self.warning_count = self._get_db().warning_count() self.rownumber = self.rownumber + len(r) return r class CursorTupleRowsMixIn: """This is a MixIn class that causes all rows to be returned as tuples, which is the standard form required by DB API.""" _fetch_type = 0 class CursorDictRowsMixIn: """This is a MixIn class that causes all rows to be returned as dictionaries. This is a non-standard feature.""" _fetch_type = 1 class Cursor(CursorStoreResultMixIn, CursorTupleRowsMixIn, BaseCursor): """This is the standard Cursor class that returns rows as tuples and stores the result set in the client.""" class DictCursor(CursorStoreResultMixIn, CursorDictRowsMixIn, BaseCursor): """This is a Cursor class that returns rows as dictionaries and stores the result set in the client.""" class SSCursor(CursorUseResultMixIn, CursorTupleRowsMixIn, BaseCursor): """This is a Cursor class that returns rows as tuples and stores the result set in the server.""" class SSDictCursor(CursorUseResultMixIn, CursorDictRowsMixIn, BaseCursor): """This is a Cursor class that returns rows as dictionaries and stores the result set in the server."""