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 "ssl_ca": "/home/gord/client-ssl/ca.pem",

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

49 "ssl_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 )

66If the server uses an automatically-generated certificate that is self-signed

67or does not match the host name (as seen from the client), it may also be

68necessary to indicate ``ssl_check_hostname=false``::

70 connection_uri = (

71 "mysql+pymysql://scott:tiger@192.168.0.134/test"

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

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

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

75 "&ssl_check_hostname=false"

76 )

79.. seealso::

81 :ref:`pymysql_ssl` in the PyMySQL dialect

84Using MySQLdb with Google Cloud SQL

85-----------------------------------

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

88using a URL like the following::

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

92Server Side Cursors

93-------------------

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

97"""

99import re

100

101from .base import MySQLCompiler

102from .base import MySQLDialect

103from .base import MySQLExecutionContext

104from .base import MySQLIdentifierPreparer

105from .base import TEXT

106from ... import sql

107from ... import util

108

109

110class MySQLExecutionContext_mysqldb(MySQLExecutionContext):

111 @property

112 def rowcount(self):

113 if hasattr(self, "_rowcount"):

114 return self._rowcount

115 else:

116 return self.cursor.rowcount

117

118

119class MySQLCompiler_mysqldb(MySQLCompiler):

120 pass

121

122

123class MySQLDialect_mysqldb(MySQLDialect):

124 driver = "mysqldb"

125 supports_statement_cache = True

126 supports_unicode_statements = True

127 supports_sane_rowcount = True

128 supports_sane_multi_rowcount = True

129

130 supports_native_decimal = True

131

132 default_paramstyle = "format"

133 execution_ctx_cls = MySQLExecutionContext_mysqldb

134 statement_compiler = MySQLCompiler_mysqldb

135 preparer = MySQLIdentifierPreparer

136

137 def __init__(self, **kwargs):

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

139 self._mysql_dbapi_version = (

140 self._parse_dbapi_version(self.dbapi.__version__)

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

142 else (0, 0, 0)

143 )

144

145 def _parse_dbapi_version(self, version):

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

147 if m:

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

149 else:

150 return (0, 0, 0)

151

152 @util.langhelpers.memoized_property

153 def supports_server_side_cursors(self):

154 try:

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

156 self._sscursor = cursors.SSCursor

157 return True

158 except (ImportError, AttributeError):

159 return False

160

161 @classmethod

162 def dbapi(cls):

163 return __import__("MySQLdb")

164

165 def on_connect(self):

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

167

168 def on_connect(conn):

169 if super_ is not None:

170 super_(conn)

171

172 charset_name = conn.character_set_name()

173

174 if charset_name is not None:

175 cursor = conn.cursor()

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

177 cursor.close()

178

179 return on_connect

180

181 def do_ping(self, dbapi_connection):

182 try:

183 dbapi_connection.ping(False)

184 except self.dbapi.Error as err:

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

186 return False

187 else:

188 raise

189 else:

190 return True

191

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

193 rowcount = cursor.executemany(statement, parameters)

194 if context is not None:

195 context._rowcount = rowcount

196

197 def _check_unicode_returns(self, connection):

198 # work around issue fixed in

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

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

201

202 collation = connection.exec_driver_sql(

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

204 % (

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

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

207 )

208 ).scalar()

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

210 if has_utf8mb4_bin:

211 additional_tests = [

212 sql.collate(

213 sql.cast(

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

215 TEXT(charset="utf8mb4"),

216 ),

217 "utf8mb4_bin",

218 )

219 ]

220 else:

221 additional_tests = []

222 return super(MySQLDialect_mysqldb, self)._check_unicode_returns(

223 connection, additional_tests

224 )

225

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

227 if _translate_args is None:

228 _translate_args = dict(

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

230 )

231

232 opts = url.translate_connect_args(**_translate_args)

233 opts.update(url.query)

234

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

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

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

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

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

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

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

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

243 # types like String and MSString.

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

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

246

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

248 # query string.

249

250 ssl = {}

251 keys = [

252 ("ssl_ca", str),

253 ("ssl_key", str),

254 ("ssl_cert", str),

255 ("ssl_capath", str),

256 ("ssl_cipher", str),

257 ("ssl_check_hostname", bool),

258 ]

259 for key, kw_type in keys:

260 if key in opts:

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

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

263 del opts[key]

264 if ssl:

265 opts["ssl"] = ssl

266

267 # FOUND_ROWS must be set in CLIENT_FLAGS to enable

268 # supports_sane_rowcount.

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

270

271 client_flag_found_rows = self._found_rows_client_flag()

272 if client_flag_found_rows is not None:

273 client_flag |= client_flag_found_rows

274 opts["client_flag"] = client_flag

275 return [[], opts]

276

277 def _found_rows_client_flag(self):

278 if self.dbapi is not None:

279 try:

280 CLIENT_FLAGS = __import__(

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

282 ).constants.CLIENT

283 except (AttributeError, ImportError):

284 return None

285 else:

286 return CLIENT_FLAGS.FOUND_ROWS

287 else:

288 return None

289

290 def _extract_error_code(self, exception):

291 return exception.args[0]

292

293 def _detect_charset(self, connection):

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

295

296 try:

297 # note: the SQL here would be

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

299 cset_name = connection.connection.character_set_name

300 except AttributeError:

301 util.warn(

302 "No 'character_set_name' can be detected with "

303 "this MySQL-Python version; "

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

305 "Assuming latin1."

306 )

307 return "latin1"

308 else:

309 return cset_name()

310

311 _isolation_lookup = set(

312 [

313 "SERIALIZABLE",

314 "READ UNCOMMITTED",

315 "READ COMMITTED",

316 "REPEATABLE READ",

317 "AUTOCOMMIT",

318 ]

319 )

320

321 def _set_isolation_level(self, connection, level):

322 if level == "AUTOCOMMIT":

323 connection.autocommit(True)

324 else:

325 connection.autocommit(False)

326 super(MySQLDialect_mysqldb, self)._set_isolation_level(

327 connection, level

328 )

329

330

331dialect = MySQLDialect_mysqldb