Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/mysql/mysqldb.py : 32%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# mysql/mysqldb.py
2# Copyright (C) 2005-2020 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: http://www.opensource.org/licenses/mit-license.php
8"""
10.. dialect:: mysql+mysqldb
11 :name: mysqlclient (maintained fork of MySQL-Python)
12 :dbapi: mysqldb
13 :connectstring: mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
14 :url: https://pypi.org/project/mysqlclient/
16Driver Status
17-------------
19The mysqlclient DBAPI is a maintained fork of the
20`MySQL-Python <http://sourceforge.net/projects/mysql-python>`_ DBAPI
21that is no longer maintained. `mysqlclient`_ supports Python 2 and Python 3
22and is very stable.
24.. _mysqlclient: https://github.com/PyMySQL/mysqlclient-python
26.. _mysqldb_unicode:
28Unicode
29-------
31Please see :ref:`mysql_unicode` for current recommendations on unicode
32handling.
35Using MySQLdb with Google Cloud SQL
36-----------------------------------
38Google Cloud SQL now recommends use of the MySQLdb dialect. Connect
39using a URL like the following::
41 mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/<projectid>:<instancename>
43Server Side Cursors
44-------------------
46The mysqldb dialect supports server-side cursors. See :ref:`mysql_ss_cursors`.
48"""
50import re
52from .base import MySQLCompiler
53from .base import MySQLDialect
54from .base import MySQLExecutionContext
55from .base import MySQLIdentifierPreparer
56from .base import TEXT
57from ... import sql
58from ... import util
61class MySQLExecutionContext_mysqldb(MySQLExecutionContext):
62 @property
63 def rowcount(self):
64 if hasattr(self, "_rowcount"):
65 return self._rowcount
66 else:
67 return self.cursor.rowcount
70class MySQLCompiler_mysqldb(MySQLCompiler):
71 pass
74class MySQLIdentifierPreparer_mysqldb(MySQLIdentifierPreparer):
75 pass
78class MySQLDialect_mysqldb(MySQLDialect):
79 driver = "mysqldb"
80 supports_unicode_statements = True
81 supports_sane_rowcount = True
82 supports_sane_multi_rowcount = True
84 supports_native_decimal = True
86 default_paramstyle = "format"
87 execution_ctx_cls = MySQLExecutionContext_mysqldb
88 statement_compiler = MySQLCompiler_mysqldb
89 preparer = MySQLIdentifierPreparer_mysqldb
91 def __init__(self, server_side_cursors=False, **kwargs):
92 super(MySQLDialect_mysqldb, self).__init__(**kwargs)
93 self.server_side_cursors = server_side_cursors
94 self._mysql_dbapi_version = (
95 self._parse_dbapi_version(self.dbapi.__version__)
96 if self.dbapi is not None and hasattr(self.dbapi, "__version__")
97 else (0, 0, 0)
98 )
100 def _parse_dbapi_version(self, version):
101 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)
102 if m:
103 return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
104 else:
105 return (0, 0, 0)
107 @util.langhelpers.memoized_property
108 def supports_server_side_cursors(self):
109 try:
110 cursors = __import__("MySQLdb.cursors").cursors
111 self._sscursor = cursors.SSCursor
112 return True
113 except (ImportError, AttributeError):
114 return False
116 @classmethod
117 def dbapi(cls):
118 return __import__("MySQLdb")
120 def on_connect(self):
121 super_ = super(MySQLDialect_mysqldb, self).on_connect()
123 def on_connect(conn):
124 if super_ is not None:
125 super_(conn)
127 charset_name = conn.character_set_name()
129 if charset_name is not None:
130 cursor = conn.cursor()
131 cursor.execute("SET NAMES %s" % charset_name)
132 cursor.close()
134 return on_connect
136 def do_ping(self, dbapi_connection):
137 try:
138 dbapi_connection.ping(False)
139 except self.dbapi.Error as err:
140 if self.is_disconnect(err, dbapi_connection, None):
141 return False
142 else:
143 raise
144 else:
145 return True
147 def do_executemany(self, cursor, statement, parameters, context=None):
148 rowcount = cursor.executemany(statement, parameters)
149 if context is not None:
150 context._rowcount = rowcount
152 def _check_unicode_returns(self, connection):
153 # work around issue fixed in
154 # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8
155 # specific issue w/ the utf8mb4_bin collation and unicode returns
157 has_utf8mb4_bin = self.server_version_info > (
158 5,
159 ) and connection.scalar(
160 "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"
161 % (
162 self.identifier_preparer.quote("Charset"),
163 self.identifier_preparer.quote("Collation"),
164 )
165 )
166 if has_utf8mb4_bin:
167 additional_tests = [
168 sql.collate(
169 sql.cast(
170 sql.literal_column("'test collated returns'"),
171 TEXT(charset="utf8mb4"),
172 ),
173 "utf8mb4_bin",
174 )
175 ]
176 else:
177 additional_tests = []
178 return super(MySQLDialect_mysqldb, self)._check_unicode_returns(
179 connection, additional_tests
180 )
182 def create_connect_args(self, url):
183 opts = url.translate_connect_args(
184 database="db", username="user", password="passwd"
185 )
186 opts.update(url.query)
188 util.coerce_kw_type(opts, "compress", bool)
189 util.coerce_kw_type(opts, "connect_timeout", int)
190 util.coerce_kw_type(opts, "read_timeout", int)
191 util.coerce_kw_type(opts, "write_timeout", int)
192 util.coerce_kw_type(opts, "client_flag", int)
193 util.coerce_kw_type(opts, "local_infile", int)
194 # Note: using either of the below will cause all strings to be
195 # returned as Unicode, both in raw SQL operations and with column
196 # types like String and MSString.
197 util.coerce_kw_type(opts, "use_unicode", bool)
198 util.coerce_kw_type(opts, "charset", str)
200 # Rich values 'cursorclass' and 'conv' are not supported via
201 # query string.
203 ssl = {}
204 keys = ["ssl_ca", "ssl_key", "ssl_cert", "ssl_capath", "ssl_cipher"]
205 for key in keys:
206 if key in opts:
207 ssl[key[4:]] = opts[key]
208 util.coerce_kw_type(ssl, key[4:], str)
209 del opts[key]
210 if ssl:
211 opts["ssl"] = ssl
213 # FOUND_ROWS must be set in CLIENT_FLAGS to enable
214 # supports_sane_rowcount.
215 client_flag = opts.get("client_flag", 0)
216 if self.dbapi is not None:
217 try:
218 CLIENT_FLAGS = __import__(
219 self.dbapi.__name__ + ".constants.CLIENT"
220 ).constants.CLIENT
221 client_flag |= CLIENT_FLAGS.FOUND_ROWS
222 except (AttributeError, ImportError):
223 self.supports_sane_rowcount = False
224 opts["client_flag"] = client_flag
225 return [[], opts]
227 def _extract_error_code(self, exception):
228 return exception.args[0]
230 def _detect_charset(self, connection):
231 """Sniff out the character set in use for connection results."""
233 try:
234 # note: the SQL here would be
235 # "SHOW VARIABLES LIKE 'character_set%%'"
236 cset_name = connection.connection.character_set_name
237 except AttributeError:
238 util.warn(
239 "No 'character_set_name' can be detected with "
240 "this MySQL-Python version; "
241 "please upgrade to a recent version of MySQL-Python. "
242 "Assuming latin1."
243 )
244 return "latin1"
245 else:
246 return cset_name()
248 _isolation_lookup = set(
249 [
250 "SERIALIZABLE",
251 "READ UNCOMMITTED",
252 "READ COMMITTED",
253 "REPEATABLE READ",
254 "AUTOCOMMIT",
255 ]
256 )
258 def _set_isolation_level(self, connection, level):
259 if level == "AUTOCOMMIT":
260 connection.autocommit(True)
261 else:
262 connection.autocommit(False)
263 super(MySQLDialect_mysqldb, self)._set_isolation_level(
264 connection, level
265 )
268dialect = MySQLDialect_mysqldb