Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pymysql/connections.py: 26%
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
1# Python implementation of the MySQL client-server protocol
2# http://dev.mysql.com/doc/internals/en/client-server-protocol.html
3# Error codes:
4# https://dev.mysql.com/doc/refman/5.5/en/error-handling.html
5import contextlib
6import errno
7import os
8import socket
9import struct
10import sys
11import traceback
12import warnings
14from . import VERSION_STRING, _auth, converters, err
15from .charset import charset_by_id, charset_by_name
16from .constants import CLIENT, COMMAND, CR, ER, FIELD_TYPE, SERVER_STATUS
17from .cursors import Cursor
18from .optionfile import Parser
19from .protocol import (
20 EOFPacketWrapper,
21 FieldDescriptorPacket,
22 LoadLocalPacketWrapper,
23 MysqlPacket,
24 OKPacketWrapper,
25 dump_packet,
26)
28try:
29 import ssl
31 SSL_ENABLED = True
32except ImportError:
33 ssl = None
34 SSL_ENABLED = False
36try:
37 import getpass
39 DEFAULT_USER = getpass.getuser()
40 del getpass
41except (ImportError, KeyError, OSError):
42 # When there's no entry in OS database for a current user:
43 # KeyError is raised in Python 3.12 and below.
44 # OSError is raised in Python 3.13+
45 DEFAULT_USER = None
47DEBUG = False
48_DEFAULT_AUTH_PLUGIN = None # if this is not None, use it instead of server's default.
50TEXT_TYPES = {
51 FIELD_TYPE.BIT,
52 FIELD_TYPE.BLOB,
53 FIELD_TYPE.LONG_BLOB,
54 FIELD_TYPE.MEDIUM_BLOB,
55 FIELD_TYPE.STRING,
56 FIELD_TYPE.TINY_BLOB,
57 FIELD_TYPE.VAR_STRING,
58 FIELD_TYPE.VARCHAR,
59 FIELD_TYPE.GEOMETRY,
60}
63DEFAULT_CHARSET = "utf8mb4"
65MAX_PACKET_LEN = 2**24 - 1
68def _pack_int24(n):
69 return struct.pack("<I", n)[:3]
72# https://dev.mysql.com/doc/internals/en/integer.html#packet-Protocol::LengthEncodedInteger
73def _lenenc_int(i):
74 if i < 0:
75 raise ValueError(
76 "Encoding %d is less than 0 - no representation in LengthEncodedInteger" % i
77 )
78 elif i < 0xFB:
79 return bytes([i])
80 elif i < (1 << 16):
81 return b"\xfc" + struct.pack("<H", i)
82 elif i < (1 << 24):
83 return b"\xfd" + struct.pack("<I", i)[:3]
84 elif i < (1 << 64):
85 return b"\xfe" + struct.pack("<Q", i)
86 else:
87 raise ValueError(
88 f"Encoding {i:x} is larger than {1 << 64:x} - no representation in LengthEncodedInteger"
89 )
92class Connection:
93 """
94 Representation of a socket with a mysql server.
96 The proper way to get an instance of this class is to call
97 connect().
99 Establish a connection to the MySQL database. Accepts several
100 arguments:
102 :param host: Host where the database server is located.
103 :param user: Username to log in as.
104 :param password: Password to use.
105 :param database: Database to use, None to not use a particular one.
106 :param port: MySQL port to use, default is usually OK. (default: 3306)
107 :param bind_address: When the client has multiple network interfaces, specify
108 the interface from which to connect to the host. Argument can be
109 a hostname or an IP address.
110 :param unix_socket: Use a unix socket rather than TCP/IP.
111 :param read_timeout: The timeout for reading from the connection in seconds.
112 (default: None - no timeout)
113 :param write_timeout: The timeout for writing to the connection in seconds.
114 (default: None - no timeout)
115 :param str charset: Charset to use.
116 :param str collation: Collation name to use.
117 :param sql_mode: Default SQL_MODE to use.
118 :param read_default_file:
119 Specifies my.cnf file to read these parameters from under the [client] section.
120 :param conv:
121 Conversion dictionary to use instead of the default one.
122 This is used to provide custom marshalling and unmarshalling of types.
123 See converters.
124 :param use_unicode:
125 Whether or not to default to unicode strings.
126 This option defaults to true.
127 :param client_flag: Custom flags to send to MySQL. Find potential values in constants.CLIENT.
128 :param cursorclass: Custom cursor class to use.
129 :param init_command: Initial SQL statement to run when connection is established.
130 :param connect_timeout: The timeout for connecting to the database in seconds.
131 (default: 10, min: 1, max: 31536000)
132 :param ssl: An ssl.SSLContext, or a dict of arguments similar to mysql_ssl_set()'s parameters.
133 Passing a dict is deprecated; use the individual ``ssl_*`` parameters or an
134 ``ssl.SSLContext`` instead.
135 :param ssl_ca: Path to the file that contains a PEM-formatted CA certificate.
136 :param ssl_cert: Path to the file that contains a PEM-formatted client certificate.
137 :param ssl_disabled: A boolean value that disables usage of TLS. Unlike other SSL options,
138 setting this to True explicitly prohibits the use of TLS, even if the server supports it.
139 :param ssl_key: Path to the file that contains a PEM-formatted private key for
140 the client certificate.
141 :param ssl_key_password: The password for the client certificate private key.
142 :param ssl_verify_cert: Set to true to check the server certificate's validity.
143 :param ssl_verify_identity: Set to true to check the server's identity.
144 :param read_default_group: Group to read from in the configuration file.
145 :param autocommit: Autocommit mode. None means use server default. (default: False)
146 :param local_infile: Boolean to enable the use of LOAD DATA LOCAL command. (default: False)
147 :param max_allowed_packet: Max size of packet sent to server in bytes. (default: 16MB)
148 Only used to limit size of "LOAD LOCAL INFILE" data packet smaller than default (16KB).
149 :param defer_connect: Don't explicitly connect on construction - wait for connect call.
150 (default: False)
151 :param auth_plugin_map: A dict of plugin names to a class that processes that plugin.
152 The class will take the Connection object as the argument to the constructor.
153 The class needs an authenticate method taking an authentication packet as
154 an argument. For the dialog plugin, a prompt(echo, prompt) method can be used
155 (if no authenticate method) for returning a string from the user. (experimental)
156 :param server_public_key: SHA256 authentication plugin public key value. (default: None)
157 :param binary_prefix: Add _binary prefix on bytes and bytearray. (default: False)
158 :param compress: Not supported.
159 :param named_pipe: Not supported.
160 :param db: **DEPRECATED** Alias for database.
161 :param passwd: **DEPRECATED** Alias for password.
163 See `Connection <https://www.python.org/dev/peps/pep-0249/#connection-objects>`_ in the
164 specification.
165 """
167 _sock = None
168 _rfile = None
169 _auth_plugin_name = ""
170 _closed = False
171 _secure = False
173 def __init__(
174 self,
175 *,
176 user=None, # The first four arguments is based on DB-API 2.0 recommendation.
177 password="",
178 host=None,
179 database=None,
180 unix_socket=None,
181 port=0,
182 charset="",
183 collation=None,
184 sql_mode=None,
185 read_default_file=None,
186 conv=None,
187 use_unicode=True,
188 client_flag=0,
189 cursorclass=Cursor,
190 init_command=None,
191 connect_timeout=10,
192 read_default_group=None,
193 autocommit=False,
194 local_infile=False,
195 max_allowed_packet=16 * 1024 * 1024,
196 defer_connect=False,
197 auth_plugin_map=None,
198 read_timeout=None,
199 write_timeout=None,
200 bind_address=None,
201 binary_prefix=False,
202 program_name=None,
203 server_public_key=None,
204 ssl=None,
205 ssl_ca=None,
206 ssl_cert=None,
207 ssl_disabled=None,
208 ssl_key=None,
209 ssl_key_password=None,
210 ssl_verify_cert=None,
211 ssl_verify_identity=None,
212 compress=None, # not supported
213 named_pipe=None, # not supported
214 passwd=None, # deprecated
215 db=None, # deprecated
216 ):
217 if db is not None and database is None:
218 warnings.warn("'db' is deprecated, use 'database'", DeprecationWarning, 3)
219 database = db
220 if passwd is not None and not password:
221 warnings.warn(
222 "'passwd' is deprecated, use 'password'", DeprecationWarning, 3
223 )
224 password = passwd
226 if compress or named_pipe:
227 raise NotImplementedError(
228 "compress and named_pipe arguments are not supported"
229 )
231 self._local_infile = bool(local_infile)
232 if self._local_infile:
233 client_flag |= CLIENT.LOCAL_FILES
235 if read_default_group and not read_default_file:
236 if sys.platform.startswith("win"):
237 read_default_file = "c:\\my.ini"
238 else:
239 read_default_file = "/etc/my.cnf"
241 if read_default_file:
242 if not read_default_group:
243 read_default_group = "client"
245 cfg = Parser()
246 cfg.read(os.path.expanduser(read_default_file))
248 def _config(key, arg):
249 if arg:
250 return arg
251 try:
252 return cfg.get(read_default_group, key)
253 except Exception:
254 return arg
256 user = _config("user", user)
257 password = _config("password", password)
258 host = _config("host", host)
259 database = _config("database", database)
260 unix_socket = _config("socket", unix_socket)
261 port = int(_config("port", port))
262 bind_address = _config("bind-address", bind_address)
263 charset = _config("default-character-set", charset)
264 if not ssl:
265 ssl = {}
266 if isinstance(ssl, dict):
267 for key in ["ca", "capath", "cert", "key", "password", "cipher"]:
268 value = _config("ssl-" + key, ssl.get(key))
269 if value:
270 ssl[key] = value
272 self.ssl = False
273 self._ssl_required = False
274 if not ssl_disabled:
275 if ssl_ca or ssl_cert or ssl_key or ssl_verify_cert or ssl_verify_identity:
276 ssl = {
277 "ca": ssl_ca,
278 "check_hostname": bool(ssl_verify_identity),
279 "verify_mode": ssl_verify_cert
280 if ssl_verify_cert is not None
281 else False,
282 }
283 if ssl_cert is not None:
284 ssl["cert"] = ssl_cert
285 if ssl_key is not None:
286 ssl["key"] = ssl_key
287 if ssl_key_password is not None:
288 ssl["password"] = ssl_key_password
289 if ssl:
290 if not SSL_ENABLED:
291 raise NotImplementedError("ssl module not found")
292 self.ssl = True
293 self._ssl_required = True
294 client_flag |= CLIENT.SSL
295 self.ctx = self._create_ssl_ctx(ssl)
296 elif SSL_ENABLED:
297 # No explicit SSL options specified: use PREFERRED mode.
298 # Attempt SSL but fall back gracefully if the server doesn't support it.
299 self.ssl = True
300 self._ssl_required = False
301 self.ctx = self._create_ssl_ctx({})
303 self.host = host or "localhost"
304 self.port = port or 3306
305 if type(self.port) is not int:
306 raise ValueError("port should be of type int")
307 self.user = user or DEFAULT_USER
308 self.password = password or b""
309 if isinstance(self.password, str):
310 self.password = self.password.encode("latin1")
311 self.db = database
312 self.unix_socket = unix_socket
313 self.bind_address = bind_address
314 if not (0 < connect_timeout <= 31536000):
315 raise ValueError("connect_timeout should be >0 and <=31536000")
316 self.connect_timeout = connect_timeout or None
317 if read_timeout is not None and read_timeout <= 0:
318 raise ValueError("read_timeout should be > 0")
319 self._read_timeout = read_timeout
320 if write_timeout is not None and write_timeout <= 0:
321 raise ValueError("write_timeout should be > 0")
322 self._write_timeout = write_timeout
324 self.charset = charset or DEFAULT_CHARSET
325 self.collation = collation
326 self.use_unicode = use_unicode
328 self.encoding = charset_by_name(self.charset).encoding
330 client_flag |= CLIENT.CAPABILITIES
331 if self.db:
332 client_flag |= CLIENT.CONNECT_WITH_DB
334 self.client_flag = client_flag
336 self.cursorclass = cursorclass
338 self._result = None
339 self._affected_rows = 0
340 self.host_info = "Not connected"
342 # specified autocommit mode. None means use server default.
343 self.autocommit_mode = autocommit
345 if conv is None:
346 conv = converters.conversions
348 # Need for MySQLdb compatibility.
349 self.encoders = {k: v for (k, v) in conv.items() if type(k) is not int}
350 self.decoders = {k: v for (k, v) in conv.items() if type(k) is int}
351 self.sql_mode = sql_mode
352 self.init_command = init_command
353 self.max_allowed_packet = max_allowed_packet
354 self._auth_plugin_map = auth_plugin_map or {}
355 self._binary_prefix = binary_prefix
356 self.server_public_key = server_public_key
358 self._connect_attrs = {
359 "_client_name": "pymysql",
360 "_client_version": VERSION_STRING,
361 "_pid": str(os.getpid()),
362 }
364 if program_name:
365 self._connect_attrs["program_name"] = program_name
367 if defer_connect:
368 self._sock = None
369 else:
370 self.connect()
372 def __enter__(self):
373 return self
375 def __exit__(self, *exc_info):
376 del exc_info
377 self.close()
379 def _create_ssl_ctx(self, sslp):
380 if isinstance(sslp, ssl.SSLContext):
381 return sslp
382 ca = sslp.get("ca")
383 capath = sslp.get("capath")
384 hasnoca = ca is None and capath is None
385 ctx = ssl.create_default_context(cafile=ca, capath=capath)
387 # Python 3.13 enables VERIFY_X509_STRICT by default.
388 # But self signed certificates that are generated by MySQL automatically
389 # doesn't pass the verification.
390 ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT
392 ctx.check_hostname = not hasnoca and sslp.get("check_hostname", True)
393 verify_mode_value = sslp.get("verify_mode")
394 if verify_mode_value is None:
395 ctx.verify_mode = ssl.CERT_NONE if hasnoca else ssl.CERT_REQUIRED
396 elif isinstance(verify_mode_value, bool):
397 ctx.verify_mode = ssl.CERT_REQUIRED if verify_mode_value else ssl.CERT_NONE
398 else:
399 if isinstance(verify_mode_value, str):
400 verify_mode_value = verify_mode_value.lower()
401 if verify_mode_value in ("none", "0", "false", "no"):
402 ctx.verify_mode = ssl.CERT_NONE
403 elif verify_mode_value == "optional":
404 ctx.verify_mode = ssl.CERT_OPTIONAL
405 elif verify_mode_value in ("required", "1", "true", "yes"):
406 ctx.verify_mode = ssl.CERT_REQUIRED
407 else:
408 ctx.verify_mode = ssl.CERT_NONE if hasnoca else ssl.CERT_REQUIRED
409 if "cert" in sslp:
410 ctx.load_cert_chain(
411 sslp["cert"], keyfile=sslp.get("key"), password=sslp.get("password")
412 )
413 if "cipher" in sslp:
414 ctx.set_ciphers(sslp["cipher"])
415 ctx.options |= ssl.OP_NO_SSLv2
416 ctx.options |= ssl.OP_NO_SSLv3
417 return ctx
419 def close(self):
420 """
421 Send the quit message and close the socket.
423 See `Connection.close() <https://www.python.org/dev/peps/pep-0249/#Connection.close>`_
424 in the specification.
426 :raise Error: If the connection is already closed.
427 """
428 if self._closed:
429 raise err.Error("Already closed")
430 self._closed = True
431 if self._sock is None:
432 return
433 send_data = struct.pack("<iB", 1, COMMAND.COM_QUIT)
434 try:
435 with contextlib.suppress(Exception):
436 self._write_bytes(send_data)
437 finally:
438 self._force_close()
440 @property
441 def open(self):
442 """Return True if the connection is open."""
443 return self._sock is not None
445 def _force_close(self):
446 """Close connection without QUIT message."""
447 if self._rfile:
448 self._rfile.close()
449 if self._sock:
450 try:
451 self._sock.close()
452 except: # noqa
453 pass
454 self._sock = None
455 self._rfile = None
457 __del__ = _force_close
459 def autocommit(self, value):
460 self.autocommit_mode = bool(value)
461 current = self.get_autocommit()
462 if value != current:
463 self._send_autocommit_mode()
465 def get_autocommit(self):
466 return bool(self.server_status & SERVER_STATUS.SERVER_STATUS_AUTOCOMMIT)
468 def _read_ok_packet(self):
469 pkt = self._read_packet()
470 if not pkt.is_ok_packet():
471 raise err.OperationalError(
472 CR.CR_COMMANDS_OUT_OF_SYNC,
473 "Command Out of Sync",
474 )
475 ok = OKPacketWrapper(pkt)
476 self.server_status = ok.server_status
477 return ok
479 def _send_autocommit_mode(self):
480 """Set whether or not to commit after every execute()."""
481 self._execute_command(
482 COMMAND.COM_QUERY, "SET AUTOCOMMIT = %s" % self.escape(self.autocommit_mode)
483 )
484 self._read_ok_packet()
486 def begin(self):
487 """Begin transaction."""
488 self._execute_command(COMMAND.COM_QUERY, "BEGIN")
489 self._read_ok_packet()
491 def commit(self):
492 """
493 Commit changes to stable storage.
495 See `Connection.commit() <https://www.python.org/dev/peps/pep-0249/#commit>`_
496 in the specification.
497 """
498 self._execute_command(COMMAND.COM_QUERY, "COMMIT")
499 self._read_ok_packet()
501 def rollback(self):
502 """
503 Roll back the current transaction.
505 See `Connection.rollback() <https://www.python.org/dev/peps/pep-0249/#rollback>`_
506 in the specification.
507 """
508 self._execute_command(COMMAND.COM_QUERY, "ROLLBACK")
509 self._read_ok_packet()
511 def show_warnings(self):
512 """Send the "SHOW WARNINGS" SQL command."""
513 self._execute_command(COMMAND.COM_QUERY, "SHOW WARNINGS")
514 result = MySQLResult(self)
515 result.read()
516 return result.rows
518 def select_db(self, db):
519 """
520 Set current db.
522 :param db: The name of the db.
523 """
524 self._execute_command(COMMAND.COM_INIT_DB, db)
525 self._read_ok_packet()
527 def escape(self, obj, mapping=None):
528 """Escape whatever value is passed.
530 Non-standard, for internal use; do not use this in your applications.
531 """
532 if isinstance(obj, str):
533 return "'" + self.escape_string(obj) + "'"
534 if isinstance(obj, (bytes, bytearray)):
535 ret = self._quote_bytes(obj)
536 if self._binary_prefix:
537 ret = "_binary" + ret
538 return ret
539 return converters.escape_item(obj, self.charset, mapping=mapping)
541 def literal(self, obj):
542 """Alias for escape().
544 Non-standard, for internal use; do not use this in your applications.
545 """
546 return self.escape(obj, self.encoders)
548 def escape_string(self, s):
549 if self.server_status & SERVER_STATUS.SERVER_STATUS_NO_BACKSLASH_ESCAPES:
550 return s.replace("'", "''")
551 return converters.escape_string(s)
553 def _quote_bytes(self, s):
554 if self.server_status & SERVER_STATUS.SERVER_STATUS_NO_BACKSLASH_ESCAPES:
555 return "'{}'".format(
556 s.replace(b"'", b"''").decode("ascii", "surrogateescape")
557 )
558 return converters.escape_bytes(s)
560 def cursor(self, cursor=None):
561 """
562 Create a new cursor to execute queries with.
564 :param cursor: The type of cursor to create. None means use Cursor.
565 :type cursor: :py:class:`Cursor`, :py:class:`SSCursor`, :py:class:`DictCursor`,
566 or :py:class:`SSDictCursor`.
567 """
568 if cursor:
569 return cursor(self)
570 return self.cursorclass(self)
572 # The following methods are INTERNAL USE ONLY (called from Cursor)
573 def query(self, sql, unbuffered=False):
574 # if DEBUG:
575 # print("DEBUG: sending query:", sql)
576 if isinstance(sql, str):
577 sql = sql.encode(self.encoding, "surrogateescape")
578 self._execute_command(COMMAND.COM_QUERY, sql)
579 self._affected_rows = self._read_query_result(unbuffered=unbuffered)
580 return self._affected_rows
582 def next_result(self, unbuffered=False):
583 self._affected_rows = self._read_query_result(unbuffered=unbuffered)
584 return self._affected_rows
586 def affected_rows(self):
587 return self._affected_rows
589 def kill(self, thread_id):
590 if not isinstance(thread_id, int):
591 raise TypeError("thread_id must be an integer")
592 self.query(f"KILL {thread_id:d}")
594 def ping(self, reconnect=False):
595 """
596 Check if the server is alive.
598 `reconnect` is deprecated. Create a new connection if you want to reconnect.
600 :param reconnect: If the connection is closed, reconnect.
601 :type reconnect: boolean
603 :raise Error: If the connection is closed and reconnect=False.
604 """
605 # emit deprecation warning for reconnect.
606 if reconnect:
607 warnings.warn(
608 "The 'reconnect' argument is deprecated. Create a new connection if you want to reconnect.",
609 DeprecationWarning,
610 2,
611 )
612 if self._sock is None:
613 if reconnect:
614 self.connect()
615 reconnect = False
616 else:
617 raise err.Error("Already closed")
618 try:
619 self._execute_command(COMMAND.COM_PING, "")
620 self._read_ok_packet()
621 except Exception:
622 if reconnect:
623 self.connect()
624 self.ping(False)
625 else:
626 raise
628 def set_charset(self, charset):
629 """Deprecated. Use set_character_set() instead."""
630 warnings.warn(
631 "'set_charset' is deprecated, use 'set_character_set' instead",
632 DeprecationWarning,
633 2,
634 )
635 # This function has been implemented in old PyMySQL.
636 # But this name is different from MySQLdb.
637 # So we keep this function for compatibility and add
638 # new set_character_set() function.
639 self.set_character_set(charset)
641 def set_character_set(self, charset, collation=None):
642 """
643 Set charset (and collation)
645 Send "SET NAMES charset [COLLATE collation]" query.
646 Update Connection.encoding based on charset.
647 """
648 # Make sure charset is supported.
649 encoding = charset_by_name(charset).encoding
651 if collation:
652 query = f"SET NAMES {charset} COLLATE {collation}"
653 else:
654 query = f"SET NAMES {charset}"
655 self._execute_command(COMMAND.COM_QUERY, query)
656 self._read_packet()
657 self.charset = charset
658 self.encoding = encoding
659 self.collation = collation
661 def connect(self, sock=None):
662 self._closed = False
663 try:
664 if sock is None:
665 if self.unix_socket:
666 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
667 sock.settimeout(self.connect_timeout)
668 sock.connect(self.unix_socket)
669 self.host_info = "Localhost via UNIX socket"
670 self._secure = True
671 if DEBUG:
672 print("connected using unix_socket")
673 else:
674 kwargs = {}
675 if self.bind_address is not None:
676 kwargs["source_address"] = (self.bind_address, 0)
677 while True:
678 try:
679 sock = socket.create_connection(
680 (self.host, self.port), self.connect_timeout, **kwargs
681 )
682 break
683 except OSError as e:
684 if e.errno == errno.EINTR:
685 continue
686 raise
687 self.host_info = "socket %s:%d" % (self.host, self.port)
688 if DEBUG:
689 print("connected using socket")
690 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
691 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
693 self._sock = sock
694 sock.settimeout(self._read_timeout)
695 self._current_timeout = self._read_timeout
696 self._rfile = sock.makefile("rb")
697 self._next_seq_id = 0
699 self._get_server_information()
700 self._request_authentication()
702 # Send "SET NAMES" query on init for:
703 # - Ensure charset (and collation) is set to the server.
704 # - collation_id in handshake packet may be ignored.
705 # - If collation is not specified, we don't know what is server's
706 # default collation for the charset. For example, default collation
707 # of utf8mb4 is:
708 # - MySQL 5.7, MariaDB 10.x: utf8mb4_general_ci
709 # - MySQL 8.0: utf8mb4_0900_ai_ci
710 #
711 # Reference:
712 # - https://github.com/PyMySQL/PyMySQL/issues/1092
713 # - https://github.com/wagtail/wagtail/issues/9477
714 # - https://zenn.dev/methane/articles/2023-mysql-collation (Japanese)
715 self.set_character_set(self.charset, self.collation)
717 if self.sql_mode is not None:
718 c = self.cursor()
719 c.execute("SET sql_mode=%s", (self.sql_mode,))
720 c.close()
722 if self.init_command is not None:
723 c = self.cursor()
724 c.execute(self.init_command)
725 c.close()
727 if self.autocommit_mode is not None:
728 self.autocommit(self.autocommit_mode)
729 except BaseException as e:
730 self._force_close()
732 if isinstance(e, OSError):
733 exc = err.OperationalError(
734 CR.CR_CONN_HOST_ERROR,
735 f"Can't connect to MySQL server on {self.host!r} ({e})",
736 )
737 # Keep original exception and traceback to investigate error.
738 exc.original_exception = e
739 exc.traceback = traceback.format_exc()
740 if DEBUG:
741 print(exc.traceback)
742 raise exc
744 # If e is neither DatabaseError or IOError, It's a bug.
745 # But raising AssertionError hides original error.
746 # So just reraise it.
747 raise
749 def write_packet(self, payload):
750 """Writes an entire "mysql packet" in its entirety to the network
751 adding its length and sequence number.
752 """
753 # Internal note: when you build packet manually and calls _write_bytes()
754 # directly, you should set self._next_seq_id properly.
755 data = _pack_int24(len(payload)) + bytes([self._next_seq_id]) + payload
756 if DEBUG:
757 dump_packet(data)
758 self._write_bytes(data)
759 self._next_seq_id = (self._next_seq_id + 1) % 256
761 def _read_packet(self, packet_type=MysqlPacket):
762 """Read an entire "mysql packet" in its entirety from the network
763 and return a MysqlPacket type that represents the results.
765 :raise OperationalError: If the connection to the MySQL server is lost.
766 :raise InternalError: If the packet sequence number is wrong.
767 """
768 # Although `socket.settimeout()` may appear fast, it temporarily releases
769 # the GIL, which can hurt performance in multithreaded applications.
770 # Avoid calling it repeatedly at high frequency.
771 if self._current_timeout != self._read_timeout:
772 self._sock.settimeout(self._read_timeout)
773 self._current_timeout = self._read_timeout
775 buff = []
776 while True:
777 packet_header = self._read_bytes(4)
778 # if DEBUG: dump_packet(packet_header)
780 btrl, btrh, packet_number = struct.unpack("<HBB", packet_header)
781 bytes_to_read = btrl + (btrh << 16)
782 if packet_number != self._next_seq_id:
783 self._force_close()
784 if packet_number == 0:
785 # MariaDB sends error packet with seqno==0 when shutdown
786 raise err.OperationalError(
787 CR.CR_SERVER_LOST,
788 "Lost connection to MySQL server during query",
789 )
790 raise err.InternalError(
791 "Packet sequence number wrong - got %d expected %d"
792 % (packet_number, self._next_seq_id)
793 )
794 self._next_seq_id = (self._next_seq_id + 1) % 256
796 recv_data = self._read_bytes(bytes_to_read)
797 if DEBUG:
798 dump_packet(recv_data)
799 buff.append(recv_data)
800 # https://dev.mysql.com/doc/internals/en/sending-more-than-16mbyte.html
801 if bytes_to_read < MAX_PACKET_LEN:
802 break
804 packet = packet_type(b"".join(buff), self.encoding)
805 if packet.is_error_packet():
806 if self._result is not None and self._result.unbuffered_active is True:
807 self._result.unbuffered_active = False
808 packet.raise_for_error()
809 return packet
811 def _read_bytes(self, num_bytes):
812 # NOTE: caller should call self._sock.settimeout(self._read_timeout)
813 # before first read.
814 while True:
815 try:
816 data = self._rfile.read(num_bytes)
817 break
818 except OSError as e:
819 if e.errno == errno.EINTR:
820 continue
821 self._force_close()
822 raise err.OperationalError(
823 CR.CR_SERVER_LOST,
824 f"Lost connection to MySQL server during query ({e})",
825 )
826 except BaseException:
827 # Don't convert unknown exception to MySQLError.
828 self._force_close()
829 raise
830 if len(data) < num_bytes:
831 self._force_close()
832 raise err.OperationalError(
833 CR.CR_SERVER_LOST, "Lost connection to MySQL server during query"
834 )
835 return data
837 def _write_bytes(self, data):
838 if self._current_timeout != self._write_timeout:
839 self._sock.settimeout(self._write_timeout)
840 self._current_timeout = self._write_timeout
841 try:
842 self._sock.sendall(data)
843 except OSError as e:
844 self._force_close()
845 raise err.OperationalError(
846 CR.CR_SERVER_GONE_ERROR, f"MySQL server has gone away ({e!r})"
847 )
849 def _read_query_result(self, unbuffered=False):
850 self._result = None
851 result = MySQLResult(self)
852 if unbuffered:
853 result.init_unbuffered_query()
854 else:
855 result.read()
856 self._result = result
857 if result.server_status is not None:
858 self.server_status = result.server_status
859 return result.affected_rows
861 def insert_id(self):
862 if self._result:
863 return self._result.insert_id
864 else:
865 return 0
867 def _execute_command(self, command, sql):
868 """
869 :raise InterfaceError: If the connection is closed.
870 :raise ValueError: If no username was specified.
871 """
872 if not self._sock:
873 raise err.InterfaceError(0, "")
875 # If the last query was unbuffered, make sure it finishes before
876 # sending new commands
877 if self._result is not None:
878 if self._result.unbuffered_active:
879 warnings.warn("Previous unbuffered result was left incomplete")
880 self._result._finish_unbuffered_query()
881 while self._result.has_next:
882 self.next_result()
883 self._result = None
885 if isinstance(sql, str):
886 sql = sql.encode(self.encoding)
888 packet_size = min(MAX_PACKET_LEN, len(sql) + 1) # +1 is for command
890 # tiny optimization: build first packet manually instead of
891 # calling self..write_packet()
892 prelude = struct.pack("<iB", packet_size, command)
893 packet = prelude + sql[: packet_size - 1]
894 self._write_bytes(packet)
895 if DEBUG:
896 dump_packet(packet)
897 self._next_seq_id = 1
899 if packet_size < MAX_PACKET_LEN:
900 return
902 sql = sql[packet_size - 1 :]
903 while True:
904 packet_size = min(MAX_PACKET_LEN, len(sql))
905 self.write_packet(sql[:packet_size])
906 sql = sql[packet_size:]
907 if not sql and packet_size < MAX_PACKET_LEN:
908 break
910 def _request_authentication(self):
911 # https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::HandshakeResponse
912 if int(self.server_version.split(".", 1)[0]) >= 5:
913 self.client_flag |= CLIENT.MULTI_RESULTS
915 if self.user is None:
916 raise ValueError("Did not specify a username")
918 charset_id = charset_by_name(self.charset).id
919 if isinstance(self.user, str):
920 self.user = self.user.encode(self.encoding)
922 # Determine flags for the initial handshake packet.
923 # CLIENT.SSL is added conditionally: for REQUIRED mode it is already set in
924 # self.client_flag, but for PREFERRED mode it is only added when the server
925 # also advertises SSL support.
926 # _do_ssl is set here and checked below for sha256_password auth.
927 client_flags = self.client_flag
928 if self.ssl:
929 if self.server_capabilities & CLIENT.SSL:
930 # SSL upgrade: include CLIENT.SSL flag and wrap the socket.
931 _do_ssl = True
932 client_flags |= CLIENT.SSL
933 elif self._ssl_required:
934 raise err.OperationalError(
935 CR.CR_SSL_CONNECTION_ERROR,
936 "SSL is required but the server doesn't support it",
937 )
938 else:
939 # PREFERRED mode: server doesn't support SSL, fall back to non-SSL.
940 _do_ssl = False
941 else:
942 _do_ssl = False
944 data_init = struct.pack(
945 "<iIB23s", client_flags, MAX_PACKET_LEN, charset_id, b""
946 )
948 if _do_ssl:
949 self.write_packet(data_init)
950 self._sock = self.ctx.wrap_socket(self._sock, server_hostname=self.host)
951 self._rfile = self._sock.makefile("rb")
952 self._secure = True
954 data = data_init + self.user + b"\0"
956 authresp = b""
957 plugin_name = None
959 if self._auth_plugin_name == "":
960 plugin_name = b""
961 authresp = _auth.scramble_native_password(self.password, self.salt)
962 elif self._auth_plugin_name == "mysql_native_password":
963 plugin_name = b"mysql_native_password"
964 authresp = _auth.scramble_native_password(self.password, self.salt)
965 elif self._auth_plugin_name == "caching_sha2_password":
966 plugin_name = b"caching_sha2_password"
967 if self.password:
968 if DEBUG:
969 print("caching_sha2: trying fast path")
970 authresp = _auth.scramble_caching_sha2(self.password, self.salt)
971 else:
972 if DEBUG:
973 print("caching_sha2: empty password")
974 elif self._auth_plugin_name == "sha256_password":
975 plugin_name = b"sha256_password"
976 if _do_ssl:
977 authresp = self.password + b"\0"
978 elif self.password:
979 authresp = b"\1" # request public key
980 else:
981 authresp = b"\0" # empty password
983 if self.server_capabilities & CLIENT.PLUGIN_AUTH_LENENC_CLIENT_DATA:
984 data += _lenenc_int(len(authresp)) + authresp
985 elif self.server_capabilities & CLIENT.SECURE_CONNECTION:
986 data += struct.pack("B", len(authresp)) + authresp
987 else: # pragma: no cover - not testing against servers without secure auth (>=5.0)
988 data += authresp + b"\0"
990 if self.db and self.server_capabilities & CLIENT.CONNECT_WITH_DB:
991 if isinstance(self.db, str):
992 self.db = self.db.encode(self.encoding)
993 data += self.db + b"\0"
995 if self.server_capabilities & CLIENT.PLUGIN_AUTH:
996 data += (plugin_name or b"") + b"\0"
998 if self.server_capabilities & CLIENT.CONNECT_ATTRS:
999 connect_attrs = b""
1000 for k, v in self._connect_attrs.items():
1001 k = k.encode("utf-8")
1002 connect_attrs += _lenenc_int(len(k)) + k
1003 v = v.encode("utf-8")
1004 connect_attrs += _lenenc_int(len(v)) + v
1005 data += _lenenc_int(len(connect_attrs)) + connect_attrs
1007 self.write_packet(data)
1008 auth_packet = self._read_packet()
1010 # Authentication is a state machine. An authentication plugin can return
1011 # another transition packet (for example, MySQL Router can request full
1012 # caching_sha2_password authentication and then switch to the backend's
1013 # mysql_native_password plugin), so keep dispatching until the server
1014 # sends a terminal packet.
1015 auth_plugin_name = self._auth_plugin_name
1016 if isinstance(auth_plugin_name, str):
1017 auth_plugin_name = auth_plugin_name.encode("ascii")
1018 auth_plugin_handler = self._get_auth_plugin_handler(auth_plugin_name)
1019 auth_switch_received = False
1021 while True:
1022 # Custom authentication handlers historically did not need to return
1023 # the final OK packet after consuming the complete exchange.
1024 if auth_packet is None and auth_plugin_handler:
1025 break
1027 # if authentication method isn't accepted the first byte
1028 # will have the octet 254
1029 if auth_packet.is_auth_switch_request():
1030 if auth_switch_received:
1031 raise err.OperationalError("received multiple auth switch requests")
1032 auth_switch_received = True
1033 if DEBUG:
1034 print("received auth switch")
1035 # https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::AuthSwitchRequest
1036 auth_packet.read_uint8() # 0xfe packet identifier
1037 plugin_name = auth_packet.read_string()
1038 if (
1039 self.server_capabilities & CLIENT.PLUGIN_AUTH
1040 and plugin_name is not None
1041 ):
1042 auth_plugin_name = plugin_name
1043 auth_plugin_handler = self._get_auth_plugin_handler(plugin_name)
1044 auth_packet = self._process_auth(
1045 plugin_name, auth_packet, auth_plugin_handler
1046 )
1047 continue
1048 raise err.OperationalError("received unknown auth switch request")
1050 if auth_packet.is_extra_auth_data():
1051 if DEBUG:
1052 print("received extra data")
1053 # https://dev.mysql.com/doc/internals/en/successful-authentication.html
1054 if auth_plugin_handler:
1055 auth_packet = auth_plugin_handler.authenticate(auth_packet)
1056 continue
1057 elif auth_plugin_name in (
1058 b"caching_sha2_password",
1059 "caching_sha2_password",
1060 ):
1061 auth_packet = _auth.caching_sha2_password_auth(self, auth_packet)
1062 continue
1063 if auth_plugin_name in (b"sha256_password", "sha256_password"):
1064 auth_packet = _auth.sha256_password_auth(self, auth_packet)
1065 continue
1066 raise err.OperationalError(
1067 "Received extra packet for auth method %r", auth_plugin_name
1068 )
1070 if auth_packet.is_ok_packet():
1071 break
1072 raise err.OperationalError("unexpected packet during authentication")
1074 if DEBUG:
1075 print("Succeed to auth")
1077 def _process_auth(self, plugin_name, auth_packet, handler=None):
1078 if handler:
1079 try:
1080 return handler.authenticate(auth_packet)
1081 except AttributeError:
1082 if plugin_name != b"dialog":
1083 raise err.OperationalError(
1084 CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
1085 f"Authentication plugin '{plugin_name}'"
1086 f" not loaded: - {type(handler)!r} missing authenticate method",
1087 )
1088 if plugin_name == b"caching_sha2_password":
1089 return _auth.caching_sha2_password_auth(self, auth_packet)
1090 elif plugin_name == b"sha256_password":
1091 return _auth.sha256_password_auth(self, auth_packet)
1092 elif plugin_name == b"mysql_native_password":
1093 data = _auth.scramble_native_password(self.password, auth_packet.read_all())
1094 elif plugin_name == b"client_ed25519":
1095 data = _auth.ed25519_password(self.password, auth_packet.read_all())
1096 elif plugin_name == b"mysql_old_password":
1097 data = (
1098 _auth.scramble_old_password(self.password, auth_packet.read_all())
1099 + b"\0"
1100 )
1101 elif plugin_name == b"mysql_clear_password":
1102 # https://dev.mysql.com/doc/internals/en/clear-text-authentication.html
1103 data = self.password + b"\0"
1104 elif plugin_name == b"dialog":
1105 pkt = auth_packet
1106 while True:
1107 flag = pkt.read_uint8()
1108 echo = (flag & 0x06) == 0x02
1109 last = (flag & 0x01) == 0x01
1110 prompt = pkt.read_all()
1112 if prompt == b"Password: ":
1113 self.write_packet(self.password + b"\0")
1114 elif handler:
1115 resp = "no response - TypeError within plugin.prompt method"
1116 try:
1117 resp = handler.prompt(echo, prompt)
1118 self.write_packet(resp + b"\0")
1119 except AttributeError:
1120 raise err.OperationalError(
1121 CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
1122 f"Authentication plugin '{plugin_name}'"
1123 f" not loaded: - {handler!r} missing prompt method",
1124 )
1125 except TypeError:
1126 raise err.OperationalError(
1127 CR.CR_AUTH_PLUGIN_ERR,
1128 f"Authentication plugin '{plugin_name}'"
1129 f" {handler!r} didn't respond with string. Returned '{resp!r}' to prompt {prompt!r}",
1130 )
1131 else:
1132 raise err.OperationalError(
1133 CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
1134 f"Authentication plugin '{plugin_name}' not configured",
1135 )
1136 pkt = self._read_packet()
1137 pkt.check_error()
1138 if pkt.is_ok_packet() or last:
1139 break
1140 return pkt
1141 else:
1142 raise err.OperationalError(
1143 CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
1144 "Authentication plugin '%s' not configured" % plugin_name,
1145 )
1147 self.write_packet(data)
1148 pkt = self._read_packet()
1149 pkt.check_error()
1150 return pkt
1152 def _get_auth_plugin_handler(self, plugin_name):
1153 plugin_class = self._auth_plugin_map.get(plugin_name)
1154 if not plugin_class and isinstance(plugin_name, bytes):
1155 plugin_class = self._auth_plugin_map.get(plugin_name.decode("ascii"))
1156 if plugin_class:
1157 try:
1158 handler = plugin_class(self)
1159 except TypeError:
1160 raise err.OperationalError(
1161 CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
1162 f"Authentication plugin '{plugin_name}'"
1163 f" not loaded: - {plugin_class!r} cannot be constructed with connection object",
1164 )
1165 else:
1166 handler = None
1167 return handler
1169 # _mysql support
1170 def thread_id(self):
1171 return self.server_thread_id[0]
1173 def character_set_name(self):
1174 return self.charset
1176 def get_host_info(self):
1177 return self.host_info
1179 def get_proto_info(self):
1180 return self.protocol_version
1182 def _get_server_information(self):
1183 i = 0
1184 packet = self._read_packet()
1185 data = packet.get_all_data()
1187 self.protocol_version = data[i]
1188 i += 1
1190 server_end = data.find(b"\0", i)
1191 self.server_version = data[i:server_end].decode("latin1")
1192 i = server_end + 1
1194 self.server_thread_id = struct.unpack("<I", data[i : i + 4])
1195 i += 4
1197 self.salt = data[i : i + 8]
1198 i += 9 # 8 + 1(filler)
1200 self.server_capabilities = struct.unpack("<H", data[i : i + 2])[0]
1201 i += 2
1203 if len(data) >= i + 6:
1204 lang, stat, cap_h, salt_len = struct.unpack("<BHHB", data[i : i + 6])
1205 i += 6
1206 # TODO: deprecate server_language and server_charset.
1207 # mysqlclient-python doesn't provide it.
1208 self.server_language = lang
1209 try:
1210 self.server_charset = charset_by_id(lang).name
1211 except KeyError:
1212 # unknown collation
1213 self.server_charset = None
1215 self.server_status = stat
1216 if DEBUG:
1217 print("server_status: %x" % stat)
1219 self.server_capabilities |= cap_h << 16
1220 if DEBUG:
1221 print("salt_len:", salt_len)
1222 salt_len = max(12, salt_len - 9)
1224 # reserved
1225 i += 10
1227 if len(data) >= i + salt_len:
1228 # salt_len includes auth_plugin_data_part_1 and filler
1229 self.salt += data[i : i + salt_len]
1230 i += salt_len
1232 i += 1
1233 # AUTH PLUGIN NAME may appear here.
1234 if self.server_capabilities & CLIENT.PLUGIN_AUTH and len(data) >= i:
1235 # Due to Bug#59453 the auth-plugin-name is missing the terminating
1236 # NUL-char in versions prior to 5.5.10 and 5.6.2.
1237 # ref: https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::Handshake
1238 # didn't use version checks as mariadb is corrected and reports
1239 # earlier than those two.
1240 server_end = data.find(b"\0", i)
1241 if server_end < 0: # pragma: no cover - very specific upstream bug
1242 # not found \0 and last field so take it all
1243 self._auth_plugin_name = data[i:].decode("utf-8")
1244 else:
1245 self._auth_plugin_name = data[i:server_end].decode("utf-8")
1247 if _DEFAULT_AUTH_PLUGIN is not None: # for tests
1248 self._auth_plugin_name = _DEFAULT_AUTH_PLUGIN
1250 def get_server_info(self):
1251 return self.server_version
1253 Warning = err.Warning
1254 Error = err.Error
1255 InterfaceError = err.InterfaceError
1256 DatabaseError = err.DatabaseError
1257 DataError = err.DataError
1258 OperationalError = err.OperationalError
1259 IntegrityError = err.IntegrityError
1260 InternalError = err.InternalError
1261 ProgrammingError = err.ProgrammingError
1262 NotSupportedError = err.NotSupportedError
1265class MySQLResult:
1266 def __init__(self, connection):
1267 """
1268 :type connection: Connection
1269 """
1270 self.connection = connection
1271 self.affected_rows = None
1272 self.insert_id = None
1273 self.server_status = None
1274 self.warning_count = 0
1275 self.message = None
1276 self.field_count = 0
1277 self.description = None
1278 self.rows = None
1279 self.has_next = None
1280 self.unbuffered_active = False
1282 def __del__(self):
1283 if self.unbuffered_active:
1284 self._finish_unbuffered_query()
1286 def read(self):
1287 try:
1288 first_packet = self.connection._read_packet()
1290 if first_packet.is_ok_packet():
1291 self._read_ok_packet(first_packet)
1292 elif first_packet.is_load_local_packet():
1293 self._read_load_local_packet(first_packet)
1294 else:
1295 self._read_result_packet(first_packet)
1296 finally:
1297 self.connection = None
1299 def init_unbuffered_query(self):
1300 """
1301 :raise OperationalError: If the connection to the MySQL server is lost.
1302 :raise InternalError:
1303 """
1304 first_packet = self.connection._read_packet()
1306 if first_packet.is_ok_packet():
1307 self.connection = None
1308 self._read_ok_packet(first_packet)
1309 elif first_packet.is_load_local_packet():
1310 try:
1311 self._read_load_local_packet(first_packet)
1312 finally:
1313 self.connection = None
1314 else:
1315 self.field_count = first_packet.read_length_encoded_integer()
1316 self._get_descriptions()
1318 # Apparently, MySQLdb picks this number because it's the maximum
1319 # value of a 64bit unsigned integer. Since we're emulating MySQLdb,
1320 # we set it to this instead of None, which would be preferred.
1321 self.affected_rows = 18446744073709551615
1322 self.unbuffered_active = True
1324 def _read_ok_packet(self, packet):
1325 ok_packet = OKPacketWrapper(packet)
1326 self.affected_rows = ok_packet.affected_rows
1327 self.insert_id = ok_packet.insert_id
1328 self.server_status = ok_packet.server_status
1329 self.warning_count = ok_packet.warning_count
1330 self.message = ok_packet.message
1331 self.has_next = ok_packet.has_next
1333 def _read_load_local_packet(self, first_packet):
1334 conn: Connection = self.connection
1335 if not conn._local_infile:
1336 raise RuntimeError(
1337 "**WARN**: Received LOAD_LOCAL packet but local_infile option is false."
1338 )
1339 load_packet = LoadLocalPacketWrapper(first_packet)
1340 try:
1341 _send_local_file(load_packet.filename, conn)
1342 finally:
1343 # send the empty packet to signify we are done sending data
1344 conn.write_packet(b"")
1345 ok_packet = conn._read_packet()
1346 # If an error occurs while sending the file, exit here without handling
1347 # the OK packet.
1349 if not ok_packet.is_ok_packet():
1350 raise err.OperationalError(
1351 CR.CR_COMMANDS_OUT_OF_SYNC, "Commands Out of Sync"
1352 )
1353 self._read_ok_packet(ok_packet)
1355 def _check_packet_is_eof(self, packet):
1356 if not packet.is_eof_packet():
1357 return False
1358 # TODO: Support CLIENT.DEPRECATE_EOF
1359 # 1) Add DEPRECATE_EOF to CAPABILITIES
1360 # 2) Mask CAPABILITIES with server_capabilities
1361 # 3) if server_capabilities & CLIENT.DEPRECATE_EOF:
1362 # use OKPacketWrapper instead of EOFPacketWrapper
1363 wp = EOFPacketWrapper(packet)
1364 self.warning_count = wp.warning_count
1365 self.has_next = wp.has_next
1366 return True
1368 def _read_result_packet(self, first_packet):
1369 self.field_count = first_packet.read_length_encoded_integer()
1370 self._get_descriptions()
1371 self._read_rowdata_packet()
1373 def _read_rowdata_packet_unbuffered(self):
1374 # Check if in an active query
1375 if not self.unbuffered_active:
1376 return
1378 # EOF
1379 packet = self.connection._read_packet()
1380 if self._check_packet_is_eof(packet):
1381 self.unbuffered_active = False
1382 self.connection = None
1383 self.rows = None
1384 return
1386 row = self._read_row_from_packet(packet)
1387 self.affected_rows = 1
1388 self.rows = (row,) # rows should tuple of row for MySQL-python compatibility.
1389 return row
1391 def _finish_unbuffered_query(self):
1392 # After much reading on the MySQL protocol, it appears that there is,
1393 # in fact, no way to stop MySQL from sending all the data after
1394 # executing a query, so we just spin, and wait for an EOF packet.
1395 while self.unbuffered_active:
1396 try:
1397 packet = self.connection._read_packet()
1398 except err.OperationalError as e:
1399 if e.args[0] in (
1400 ER.QUERY_TIMEOUT,
1401 ER.STATEMENT_TIMEOUT,
1402 ):
1403 # if the query timed out we can simply ignore this error
1404 self.unbuffered_active = False
1405 self.connection = None
1406 return
1408 raise
1410 if self._check_packet_is_eof(packet):
1411 self.unbuffered_active = False
1412 self.connection = None # release reference to kill cyclic reference.
1414 def _read_rowdata_packet(self):
1415 """Read a rowdata packet for each data row in the result set."""
1416 rows = []
1417 while True:
1418 packet = self.connection._read_packet()
1419 if self._check_packet_is_eof(packet):
1420 self.connection = None # release reference to kill cyclic reference.
1421 break
1422 rows.append(self._read_row_from_packet(packet))
1424 self.affected_rows = len(rows)
1425 self.rows = tuple(rows)
1427 def _read_row_from_packet(self, packet):
1428 row = []
1429 for encoding, converter in self.converters:
1430 try:
1431 data = packet.read_length_coded_string()
1432 except IndexError:
1433 # No more columns in this row
1434 # See https://github.com/PyMySQL/PyMySQL/pull/434
1435 break
1436 if data is not None:
1437 if encoding is not None:
1438 data = data.decode(encoding)
1439 if DEBUG:
1440 print("DEBUG: DATA = ", data)
1441 if converter is not None:
1442 data = converter(data)
1443 row.append(data)
1444 return tuple(row)
1446 def _get_descriptions(self):
1447 """Read a column descriptor packet for each column in the result."""
1448 self.fields = []
1449 self.converters = []
1450 use_unicode = self.connection.use_unicode
1451 conn_encoding = self.connection.encoding
1452 description = []
1454 for i in range(self.field_count):
1455 field = self.connection._read_packet(FieldDescriptorPacket)
1456 self.fields.append(field)
1457 description.append(field.description())
1458 field_type = field.type_code
1459 if use_unicode:
1460 if field_type == FIELD_TYPE.JSON:
1461 # When SELECT from JSON column: charset = binary
1462 # When SELECT CAST(... AS JSON): charset = connection encoding
1463 # This behavior is different from TEXT / BLOB.
1464 # We should decode result by connection encoding regardless charsetnr.
1465 # See https://github.com/PyMySQL/PyMySQL/issues/488
1466 encoding = conn_encoding # SELECT CAST(... AS JSON)
1467 elif field_type in TEXT_TYPES:
1468 if field.charsetnr == 63: # binary
1469 # TEXTs with charset=binary means BINARY types.
1470 encoding = None
1471 else:
1472 encoding = conn_encoding
1473 else:
1474 # Integers, Dates and Times, and other basic data is encoded in ascii
1475 encoding = "ascii"
1476 else:
1477 encoding = None
1478 converter = self.connection.decoders.get(field_type)
1479 if converter is converters.through:
1480 converter = None
1481 if DEBUG:
1482 print(f"DEBUG: field={field}, converter={converter}")
1483 self.converters.append((encoding, converter))
1485 eof_packet = self.connection._read_packet()
1486 assert eof_packet.is_eof_packet(), "Protocol error, expecting EOF"
1487 self.description = tuple(description)
1490def _send_local_file(filename: str, conn: Connection):
1491 """Send data packets from the local file to the server"""
1492 packet_size = min(conn.max_allowed_packet, 16 * 1024)
1494 try:
1495 with open(filename, "rb") as file:
1496 # 16KB is efficient enough
1497 while True:
1498 chunk = file.read(packet_size)
1499 if not chunk:
1500 break
1501 conn.write_packet(chunk)
1502 except OSError as e:
1503 raise err.OperationalError(
1504 ER.FILE_NOT_FOUND,
1505 f"Can't open file '{filename}': {e}",
1506 )