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

1# dialects/mysql/mysqldb.py 

2# Copyright (C) 2005-2025 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.. sourcecode:: text 

78 

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

80 

81Server Side Cursors 

82------------------- 

83 

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

85 

86""" 

87from __future__ import annotations 

88 

89import re 

90from typing import Any 

91from typing import Callable 

92from typing import cast 

93from typing import Dict 

94from typing import Optional 

95from typing import Tuple 

96from typing import TYPE_CHECKING 

97 

98from .base import MySQLCompiler 

99from .base import MySQLDialect 

100from .base import MySQLExecutionContext 

101from .base import MySQLIdentifierPreparer 

102from ... import util 

103from ...util.typing import Literal 

104 

105if TYPE_CHECKING: 

106 

107 from ...engine.base import Connection 

108 from ...engine.interfaces import _DBAPIMultiExecuteParams 

109 from ...engine.interfaces import ConnectArgsType 

110 from ...engine.interfaces import DBAPIConnection 

111 from ...engine.interfaces import DBAPICursor 

112 from ...engine.interfaces import DBAPIModule 

113 from ...engine.interfaces import ExecutionContext 

114 from ...engine.interfaces import IsolationLevel 

115 from ...engine.url import URL 

116 

117 

118class MySQLExecutionContext_mysqldb(MySQLExecutionContext): 

119 pass 

120 

121 

122class MySQLCompiler_mysqldb(MySQLCompiler): 

123 pass 

124 

125 

126class MySQLDialect_mysqldb(MySQLDialect): 

127 driver = "mysqldb" 

128 supports_statement_cache = True 

129 supports_unicode_statements = True 

130 supports_sane_rowcount = True 

131 supports_sane_multi_rowcount = True 

132 

133 supports_native_decimal = True 

134 

135 default_paramstyle = "format" 

136 execution_ctx_cls = MySQLExecutionContext_mysqldb 

137 statement_compiler = MySQLCompiler_mysqldb 

138 preparer = MySQLIdentifierPreparer 

139 server_version_info: Tuple[int, ...] 

140 

141 def __init__(self, **kwargs: Any): 

142 super().__init__(**kwargs) 

143 self._mysql_dbapi_version = ( 

144 self._parse_dbapi_version(self.dbapi.__version__) 

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

146 else (0, 0, 0) 

147 ) 

148 

149 def _parse_dbapi_version(self, version: str) -> Tuple[int, ...]: 

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

151 if m: 

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

153 else: 

154 return (0, 0, 0) 

155 

156 @util.langhelpers.memoized_property 

157 def supports_server_side_cursors(self) -> bool: 

158 try: 

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

160 self._sscursor = cursors.SSCursor 

161 return True 

162 except (ImportError, AttributeError): 

163 return False 

164 

165 @classmethod 

166 def import_dbapi(cls) -> DBAPIModule: 

167 return __import__("MySQLdb") 

168 

169 def on_connect(self) -> Callable[[DBAPIConnection], None]: 

170 super_ = super().on_connect() 

171 

172 def on_connect(conn: DBAPIConnection) -> None: 

173 if super_ is not None: 

174 super_(conn) 

175 

176 charset_name = conn.character_set_name() 

177 

178 if charset_name is not None: 

179 cursor = conn.cursor() 

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

181 cursor.close() 

182 

183 return on_connect 

184 

185 def do_ping(self, dbapi_connection: DBAPIConnection) -> Literal[True]: 

186 dbapi_connection.ping() 

187 return True 

188 

189 def do_executemany( 

190 self, 

191 cursor: DBAPICursor, 

192 statement: str, 

193 parameters: _DBAPIMultiExecuteParams, 

194 context: Optional[ExecutionContext] = None, 

195 ) -> None: 

196 rowcount = cursor.executemany(statement, parameters) 

197 if context is not None: 

198 cast(MySQLExecutionContext, context)._rowcount = rowcount 

199 

200 def create_connect_args( 

201 self, url: URL, _translate_args: Optional[Dict[str, Any]] = None 

202 ) -> ConnectArgsType: 

203 if _translate_args is None: 

204 _translate_args = dict( 

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

206 ) 

207 

208 opts = url.translate_connect_args(**_translate_args) 

209 opts.update(url.query) 

210 

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

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

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

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

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

216 util.coerce_kw_type(opts, "local_infile", bool) 

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

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

219 # types like String and MSString. 

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

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

222 

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

224 # query string. 

225 

226 ssl = {} 

227 keys = [ 

228 ("ssl_ca", str), 

229 ("ssl_key", str), 

230 ("ssl_cert", str), 

231 ("ssl_capath", str), 

232 ("ssl_cipher", str), 

233 ("ssl_check_hostname", bool), 

234 ] 

235 for key, kw_type in keys: 

236 if key in opts: 

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

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

239 del opts[key] 

240 if ssl: 

241 opts["ssl"] = ssl 

242 

243 # FOUND_ROWS must be set in CLIENT_FLAGS to enable 

244 # supports_sane_rowcount. 

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

246 

247 client_flag_found_rows = self._found_rows_client_flag() 

248 if client_flag_found_rows is not None: 

249 client_flag |= client_flag_found_rows 

250 opts["client_flag"] = client_flag 

251 return [], opts 

252 

253 def _found_rows_client_flag(self) -> Optional[int]: 

254 if self.dbapi is not None: 

255 try: 

256 CLIENT_FLAGS = __import__( 

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

258 ).constants.CLIENT 

259 except (AttributeError, ImportError): 

260 return None 

261 else: 

262 return CLIENT_FLAGS.FOUND_ROWS # type: ignore 

263 else: 

264 return None 

265 

266 def _extract_error_code(self, exception: DBAPIModule.Error) -> int: 

267 return exception.args[0] # type: ignore[no-any-return] 

268 

269 def _detect_charset(self, connection: Connection) -> str: 

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

271 

272 try: 

273 # note: the SQL here would be 

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

275 

276 cset_name: Callable[[], str] = ( 

277 connection.connection.character_set_name 

278 ) 

279 except AttributeError: 

280 util.warn( 

281 "No 'character_set_name' can be detected with " 

282 "this MySQL-Python version; " 

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

284 "Assuming latin1." 

285 ) 

286 return "latin1" 

287 else: 

288 return cset_name() 

289 

290 def get_isolation_level_values( 

291 self, dbapi_conn: DBAPIConnection 

292 ) -> Tuple[IsolationLevel, ...]: 

293 return ( 

294 "SERIALIZABLE", 

295 "READ UNCOMMITTED", 

296 "READ COMMITTED", 

297 "REPEATABLE READ", 

298 "AUTOCOMMIT", 

299 ) 

300 

301 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool: 

302 return dbapi_conn.get_autocommit() # type: ignore[no-any-return] 

303 

304 def set_isolation_level( 

305 self, dbapi_connection: DBAPIConnection, level: IsolationLevel 

306 ) -> None: 

307 if level == "AUTOCOMMIT": 

308 dbapi_connection.autocommit(True) 

309 else: 

310 dbapi_connection.autocommit(False) 

311 super().set_isolation_level(dbapi_connection, level) 

312 

313 

314dialect = MySQLDialect_mysqldb