Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/mysqldb.py: 33%

1# mysql/mysqldb.py

3# <see AUTHORS file>

5# This module is part of SQLAlchemy and is released under

6# the MIT License: https://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 <https://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.

34.. _mysqldb_ssl:

36SSL Connections

37----------------

39The mysqlclient and PyMySQL DBAPIs accept an additional dictionary under the

40key "ssl", which may be specified using the

41:paramref:`_sa.create_engine.connect_args` dictionary::

43 engine = create_engine(

44 "mysql+mysqldb://scott:tiger@192.168.0.134/test",

45 connect_args={

46 "ssl": {

47 "ca": "/home/gord/client-ssl/ca.pem",

48 "cert": "/home/gord/client-ssl/client-cert.pem",

49 "key": "/home/gord/client-ssl/client-key.pem"

50 }

51 }

52 )

54For convenience, the following keys may also be specified inline within the URL

55where they will be interpreted into the "ssl" dictionary automatically:

56"ssl_ca", "ssl_cert", "ssl_key", "ssl_capath", "ssl_cipher",

57"ssl_check_hostname". An example is as follows::

59 connection_uri = (

60 "mysql+mysqldb://scott:tiger@192.168.0.134/test"

61 "?ssl_ca=/home/gord/client-ssl/ca.pem"

62 "&ssl_cert=/home/gord/client-ssl/client-cert.pem"

63 "&ssl_key=/home/gord/client-ssl/client-key.pem"

64 )

66.. seealso::

68 :ref:`pymysql_ssl` in the PyMySQL dialect

71Using MySQLdb with Google Cloud SQL

72-----------------------------------

74Google Cloud SQL now recommends use of the MySQLdb dialect. Connect

75using a URL like the following::

77 mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/<projectid>:<instancename>

79Server Side Cursors

80-------------------

82The mysqldb dialect supports server-side cursors. See :ref:`mysql_ss_cursors`.

84"""

86import re

88from .base import MySQLCompiler

89from .base import MySQLDialect

90from .base import MySQLExecutionContext

91from .base import MySQLIdentifierPreparer

92from .base import TEXT

93from ... import sql

94from ... import util

97class MySQLExecutionContext_mysqldb(MySQLExecutionContext):

98 @property

99 def rowcount(self):

100 if hasattr(self, "_rowcount"):

101 return self._rowcount

102 else:

103 return self.cursor.rowcount

104

105

106class MySQLCompiler_mysqldb(MySQLCompiler):

107 pass

108

109

110class MySQLDialect_mysqldb(MySQLDialect):

111 driver = "mysqldb"

112 supports_statement_cache = True

113 supports_unicode_statements = True

114 supports_sane_rowcount = True

115 supports_sane_multi_rowcount = True

116

117 supports_native_decimal = True

118

119 default_paramstyle = "format"

120 execution_ctx_cls = MySQLExecutionContext_mysqldb

121 statement_compiler = MySQLCompiler_mysqldb

122 preparer = MySQLIdentifierPreparer

123

124 def __init__(self, **kwargs):

125 super(MySQLDialect_mysqldb, self).__init__(**kwargs)

126 self._mysql_dbapi_version = (

127 self._parse_dbapi_version(self.dbapi.__version__)

128 if self.dbapi is not None and hasattr(self.dbapi, "__version__")

129 else (0, 0, 0)

130 )

131

132 def _parse_dbapi_version(self, version):

133 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)

134 if m:

135 return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)

136 else:

137 return (0, 0, 0)

138

139 @util.langhelpers.memoized_property

140 def supports_server_side_cursors(self):

141 try:

142 cursors = __import__("MySQLdb.cursors").cursors

143 self._sscursor = cursors.SSCursor

144 return True

145 except (ImportError, AttributeError):

146 return False

147

148 @classmethod

149 def dbapi(cls):

150 return __import__("MySQLdb")

151

152 def on_connect(self):

153 super_ = super(MySQLDialect_mysqldb, self).on_connect()

154

155 def on_connect(conn):

156 if super_ is not None:

157 super_(conn)

158

159 charset_name = conn.character_set_name()

160

161 if charset_name is not None:

162 cursor = conn.cursor()

163 cursor.execute("SET NAMES %s" % charset_name)

164 cursor.close()

165

166 return on_connect

167

168 def do_ping(self, dbapi_connection):

169 try:

170 dbapi_connection.ping(False)

171 except self.dbapi.Error as err:

172 if self.is_disconnect(err, dbapi_connection, None):

173 return False

174 else:

175 raise

176 else:

177 return True

178

179 def do_executemany(self, cursor, statement, parameters, context=None):

180 rowcount = cursor.executemany(statement, parameters)

181 if context is not None:

182 context._rowcount = rowcount

183

184 def _check_unicode_returns(self, connection):

185 # work around issue fixed in

186 # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8

187 # specific issue w/ the utf8mb4_bin collation and unicode returns

188

189 collation = connection.exec_driver_sql(

190 "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"

191 % (

192 self.identifier_preparer.quote("Charset"),

193 self.identifier_preparer.quote("Collation"),

194 )

195 ).scalar()

196 has_utf8mb4_bin = self.server_version_info > (5,) and collation

197 if has_utf8mb4_bin:

198 additional_tests = [

199 sql.collate(

200 sql.cast(

201 sql.literal_column("'test collated returns'"),

202 TEXT(charset="utf8mb4"),

203 ),

204 "utf8mb4_bin",

205 )

206 ]

207 else:

208 additional_tests = []

209 return super(MySQLDialect_mysqldb, self)._check_unicode_returns(

210 connection, additional_tests

211 )

212

213 def create_connect_args(self, url, _translate_args=None):

214 if _translate_args is None:

215 _translate_args = dict(

216 database="db", username="user", password="passwd"

217 )

218

219 opts = url.translate_connect_args(**_translate_args)

220 opts.update(url.query)

221

222 util.coerce_kw_type(opts, "compress", bool)

223 util.coerce_kw_type(opts, "connect_timeout", int)

224 util.coerce_kw_type(opts, "read_timeout", int)

225 util.coerce_kw_type(opts, "write_timeout", int)

226 util.coerce_kw_type(opts, "client_flag", int)

227 util.coerce_kw_type(opts, "local_infile", int)

228 # Note: using either of the below will cause all strings to be

229 # returned as Unicode, both in raw SQL operations and with column

230 # types like String and MSString.

231 util.coerce_kw_type(opts, "use_unicode", bool)

232 util.coerce_kw_type(opts, "charset", str)

233

234 # Rich values 'cursorclass' and 'conv' are not supported via

235 # query string.

236

237 ssl = {}

238 keys = [

239 ("ssl_ca", str),

240 ("ssl_key", str),

241 ("ssl_cert", str),

242 ("ssl_capath", str),

243 ("ssl_cipher", str),

244 ("ssl_check_hostname", bool),

245 ]

246 for key, kw_type in keys:

247 if key in opts:

248 ssl[key[4:]] = opts[key]

249 util.coerce_kw_type(ssl, key[4:], kw_type)

250 del opts[key]

251 if ssl:

252 opts["ssl"] = ssl

253

254 # FOUND_ROWS must be set in CLIENT_FLAGS to enable

255 # supports_sane_rowcount.

256 client_flag = opts.get("client_flag", 0)

257

258 client_flag_found_rows = self._found_rows_client_flag()

259 if client_flag_found_rows is not None:

260 client_flag |= client_flag_found_rows

261 opts["client_flag"] = client_flag

262 return [[], opts]

263

264 def _found_rows_client_flag(self):

265 if self.dbapi is not None:

266 try:

267 CLIENT_FLAGS = __import__(

268 self.dbapi.__name__ + ".constants.CLIENT"

269 ).constants.CLIENT

270 except (AttributeError, ImportError):

271 return None

272 else:

273 return CLIENT_FLAGS.FOUND_ROWS

274 else:

275 return None

276

277 def _extract_error_code(self, exception):

278 return exception.args[0]

279

280 def _detect_charset(self, connection):

281 """Sniff out the character set in use for connection results."""

282

283 try:

284 # note: the SQL here would be

285 # "SHOW VARIABLES LIKE 'character_set%%'"

286 cset_name = connection.connection.character_set_name

287 except AttributeError:

288 util.warn(

289 "No 'character_set_name' can be detected with "

290 "this MySQL-Python version; "

291 "please upgrade to a recent version of MySQL-Python. "

292 "Assuming latin1."

293 )

294 return "latin1"

295 else:

296 return cset_name()

297

298 _isolation_lookup = set(

299 [

300 "SERIALIZABLE",

301 "READ UNCOMMITTED",

302 "READ COMMITTED",

303 "REPEATABLE READ",

304 "AUTOCOMMIT",

305 ]

306 )

307

308 def _set_isolation_level(self, connection, level):

309 if level == "AUTOCOMMIT":

310 connection.autocommit(True)

311 else:

312 connection.autocommit(False)

313 super(MySQLDialect_mysqldb, self)._set_isolation_level(

314 connection, level

315 )

316

317

318dialect = MySQLDialect_mysqldb