1# dialects/mysql/mysqlconnector.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7
8r"""
9.. dialect:: mysql+mysqlconnector
10 :name: MySQL Connector/Python
11 :dbapi: myconnpy
12 :connectstring: mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
13 :url: https://pypi.org/project/mysql-connector-python/
14
15.. note::
16
17 The MySQL Connector/Python DBAPI has had many issues since its release,
18 some of which may remain unresolved, and the mysqlconnector dialect is
19 **not tested as part of SQLAlchemy's continuous integration**.
20 The recommended MySQL dialects are mysqlclient and PyMySQL.
21
22""" # noqa
23
24import re
25
26from .base import BIT
27from .base import MySQLCompiler
28from .base import MySQLDialect
29from .base import MySQLIdentifierPreparer
30from ... import processors
31from ... import util
32
33
34class MySQLCompiler_mysqlconnector(MySQLCompiler):
35 def visit_mod_binary(self, binary, operator, **kw):
36 if self.dialect._mysqlconnector_double_percents:
37 return (
38 self.process(binary.left, **kw)
39 + " %% "
40 + self.process(binary.right, **kw)
41 )
42 else:
43 return (
44 self.process(binary.left, **kw)
45 + " % "
46 + self.process(binary.right, **kw)
47 )
48
49 def post_process_text(self, text):
50 if self.dialect._mysqlconnector_double_percents:
51 return text.replace("%", "%%")
52 else:
53 return text
54
55 def escape_literal_column(self, text):
56 if self.dialect._mysqlconnector_double_percents:
57 return text.replace("%", "%%")
58 else:
59 return text
60
61
62class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
63 @property
64 def _double_percents(self):
65 return self.dialect._mysqlconnector_double_percents
66
67 @_double_percents.setter
68 def _double_percents(self, value):
69 pass
70
71 def _escape_identifier(self, value):
72 value = value.replace(self.escape_quote, self.escape_to_quote)
73 if self.dialect._mysqlconnector_double_percents:
74 return value.replace("%", "%%")
75 else:
76 return value
77
78
79class _myconnpyBIT(BIT):
80 def result_processor(self, dialect, coltype):
81 """MySQL-connector already converts mysql bits, so."""
82
83 return None
84
85
86class MySQLDialect_mysqlconnector(MySQLDialect):
87 driver = "mysqlconnector"
88 supports_statement_cache = True
89
90 supports_unicode_binds = True
91
92 supports_sane_rowcount = True
93 supports_sane_multi_rowcount = True
94
95 supports_native_decimal = True
96
97 default_paramstyle = "format"
98 statement_compiler = MySQLCompiler_mysqlconnector
99
100 preparer = MySQLIdentifierPreparer_mysqlconnector
101
102 colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _myconnpyBIT})
103
104 def __init__(self, *arg, **kw):
105 super(MySQLDialect_mysqlconnector, self).__init__(*arg, **kw)
106
107 # hack description encoding since mysqlconnector randomly
108 # returns bytes or not
109 self._description_decoder = (
110 processors.to_conditional_unicode_processor_factory
111 )(self.description_encoding)
112
113 def _check_unicode_description(self, connection):
114 # hack description encoding since mysqlconnector randomly
115 # returns bytes or not
116 return False
117
118 @property
119 def description_encoding(self):
120 # total guess
121 return "latin-1"
122
123 @util.memoized_property
124 def supports_unicode_statements(self):
125 return util.py3k or self._mysqlconnector_version_info > (2, 0)
126
127 @classmethod
128 def dbapi(cls):
129 from mysql import connector
130
131 return connector
132
133 def do_ping(self, dbapi_connection):
134 try:
135 dbapi_connection.ping(False)
136 except self.dbapi.Error as err:
137 if self.is_disconnect(err, dbapi_connection, None):
138 return False
139 else:
140 raise
141 else:
142 return True
143
144 def create_connect_args(self, url):
145 opts = url.translate_connect_args(username="user")
146
147 opts.update(url.query)
148
149 util.coerce_kw_type(opts, "allow_local_infile", bool)
150 util.coerce_kw_type(opts, "autocommit", bool)
151 util.coerce_kw_type(opts, "buffered", bool)
152 util.coerce_kw_type(opts, "compress", bool)
153 util.coerce_kw_type(opts, "connection_timeout", int)
154 util.coerce_kw_type(opts, "connect_timeout", int)
155 util.coerce_kw_type(opts, "consume_results", bool)
156 util.coerce_kw_type(opts, "force_ipv6", bool)
157 util.coerce_kw_type(opts, "get_warnings", bool)
158 util.coerce_kw_type(opts, "pool_reset_session", bool)
159 util.coerce_kw_type(opts, "pool_size", int)
160 util.coerce_kw_type(opts, "raise_on_warnings", bool)
161 util.coerce_kw_type(opts, "raw", bool)
162 util.coerce_kw_type(opts, "ssl_verify_cert", bool)
163 util.coerce_kw_type(opts, "use_pure", bool)
164 util.coerce_kw_type(opts, "use_unicode", bool)
165
166 # unfortunately, MySQL/connector python refuses to release a
167 # cursor without reading fully, so non-buffered isn't an option
168 opts.setdefault("buffered", True)
169
170 # FOUND_ROWS must be set in ClientFlag to enable
171 # supports_sane_rowcount.
172 if self.dbapi is not None:
173 try:
174 from mysql.connector.constants import ClientFlag
175
176 client_flags = opts.get(
177 "client_flags", ClientFlag.get_default()
178 )
179 client_flags |= ClientFlag.FOUND_ROWS
180 opts["client_flags"] = client_flags
181 except Exception:
182 pass
183 return [[], opts]
184
185 @util.memoized_property
186 def _mysqlconnector_version_info(self):
187 if self.dbapi and hasattr(self.dbapi, "__version__"):
188 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
189 if m:
190 return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
191
192 @util.memoized_property
193 def _mysqlconnector_double_percents(self):
194 return not util.py3k and self._mysqlconnector_version_info < (2, 0)
195
196 def _detect_charset(self, connection):
197 return connection.connection.charset
198
199 def _extract_error_code(self, exception):
200 return exception.errno
201
202 def is_disconnect(self, e, connection, cursor):
203 errnos = (2006, 2013, 2014, 2045, 2055, 2048)
204 exceptions = (self.dbapi.OperationalError, self.dbapi.InterfaceError)
205 if isinstance(e, exceptions):
206 return (
207 e.errno in errnos
208 or "MySQL Connection not available." in str(e)
209 or "Connection to MySQL is not available" in str(e)
210 )
211 else:
212 return False
213
214 def _compat_fetchall(self, rp, charset=None):
215 return rp.fetchall()
216
217 def _compat_fetchone(self, rp, charset=None):
218 return rp.fetchone()
219
220 _isolation_lookup = set(
221 [
222 "SERIALIZABLE",
223 "READ UNCOMMITTED",
224 "READ COMMITTED",
225 "REPEATABLE READ",
226 "AUTOCOMMIT",
227 ]
228 )
229
230 def _set_isolation_level(self, connection, level):
231 if level == "AUTOCOMMIT":
232 connection.autocommit = True
233 else:
234 connection.autocommit = False
235 super(MySQLDialect_mysqlconnector, self)._set_isolation_level(
236 connection, level
237 )
238
239
240dialect = MySQLDialect_mysqlconnector