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

1# dialects/mysql/mysqldb.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 

8""" 

9 

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/ 

15 

16Driver Status 

17------------- 

18 

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. 

23 

24.. _mysqlclient: https://github.com/PyMySQL/mysqlclient-python 

25 

26.. _mysqldb_unicode: 

27 

28Unicode 

29------- 

30 

31Please see :ref:`mysql_unicode` for current recommendations on unicode 

32handling. 

33 

34.. _mysqldb_ssl: 

35 

36SSL Connections 

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

38 

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:: 

42 

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 ) 

53 

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:: 

58 

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 ) 

65 

66.. seealso:: 

67 

68 :ref:`pymysql_ssl` in the PyMySQL dialect 

69 

70 

71Using MySQLdb with Google Cloud SQL 

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

73 

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

75using a URL like the following:: 

76 

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

78 

79Server Side Cursors 

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

81 

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

83 

84""" 

85 

86import re 

87 

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 

95 

96 

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 _ping_impl(self, dbapi_connection): 

169 return dbapi_connection.ping() 

170 

171 def do_ping(self, dbapi_connection): 

172 try: 

173 self._ping_impl(dbapi_connection) 

174 except self.dbapi.Error as err: 

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

176 return False 

177 else: 

178 raise 

179 else: 

180 return True 

181 

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

183 rowcount = cursor.executemany(statement, parameters) 

184 if context is not None: 

185 context._rowcount = rowcount 

186 

187 def _check_unicode_returns(self, connection): 

188 # work around issue fixed in 

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

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

191 

192 collation = connection.exec_driver_sql( 

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

194 % ( 

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

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

197 ) 

198 ).scalar() 

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

200 if has_utf8mb4_bin: 

201 additional_tests = [ 

202 sql.collate( 

203 sql.cast( 

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

205 TEXT(charset="utf8mb4"), 

206 ), 

207 "utf8mb4_bin", 

208 ) 

209 ] 

210 else: 

211 additional_tests = [] 

212 return super(MySQLDialect_mysqldb, self)._check_unicode_returns( 

213 connection, additional_tests 

214 ) 

215 

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

217 if _translate_args is None: 

218 _translate_args = dict( 

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

220 ) 

221 

222 opts = url.translate_connect_args(**_translate_args) 

223 opts.update(url.query) 

224 

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

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

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

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

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

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

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

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

233 # types like String and MSString. 

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

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

236 

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

238 # query string. 

239 

240 ssl = {} 

241 keys = [ 

242 ("ssl_ca", str), 

243 ("ssl_key", str), 

244 ("ssl_cert", str), 

245 ("ssl_capath", str), 

246 ("ssl_cipher", str), 

247 ("ssl_check_hostname", bool), 

248 ] 

249 for key, kw_type in keys: 

250 if key in opts: 

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

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

253 del opts[key] 

254 if ssl: 

255 opts["ssl"] = ssl 

256 

257 # FOUND_ROWS must be set in CLIENT_FLAGS to enable 

258 # supports_sane_rowcount. 

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

260 

261 client_flag_found_rows = self._found_rows_client_flag() 

262 if client_flag_found_rows is not None: 

263 client_flag |= client_flag_found_rows 

264 opts["client_flag"] = client_flag 

265 return [[], opts] 

266 

267 def _found_rows_client_flag(self): 

268 if self.dbapi is not None: 

269 try: 

270 CLIENT_FLAGS = __import__( 

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

272 ).constants.CLIENT 

273 except (AttributeError, ImportError): 

274 return None 

275 else: 

276 return CLIENT_FLAGS.FOUND_ROWS 

277 else: 

278 return None 

279 

280 def _extract_error_code(self, exception): 

281 return exception.args[0] 

282 

283 def _detect_charset(self, connection): 

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

285 

286 try: 

287 # note: the SQL here would be 

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

289 cset_name = connection.connection.character_set_name 

290 except AttributeError: 

291 util.warn( 

292 "No 'character_set_name' can be detected with " 

293 "this MySQL-Python version; " 

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

295 "Assuming latin1." 

296 ) 

297 return "latin1" 

298 else: 

299 return cset_name() 

300 

301 _isolation_lookup = set( 

302 [ 

303 "SERIALIZABLE", 

304 "READ UNCOMMITTED", 

305 "READ COMMITTED", 

306 "REPEATABLE READ", 

307 "AUTOCOMMIT", 

308 ] 

309 ) 

310 

311 def _set_isolation_level(self, connection, level): 

312 if level == "AUTOCOMMIT": 

313 connection.autocommit(True) 

314 else: 

315 connection.autocommit(False) 

316 super(MySQLDialect_mysqldb, self)._set_isolation_level( 

317 connection, level 

318 ) 

319 

320 

321dialect = MySQLDialect_mysqldb