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

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

9r""" 

10.. dialect:: mysql+mysqlconnector 

11 :name: MySQL Connector/Python 

12 :dbapi: myconnpy 

13 :connectstring: mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> 

14 :url: https://pypi.org/project/mysql-connector-python/ 

15 

16Driver Status 

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

18 

19MySQL Connector/Python is supported as of SQLAlchemy 2.0.39 to the 

20degree which the driver is functional. There are still ongoing issues 

21with features such as server side cursors which remain disabled until 

22upstream issues are repaired. 

23 

24.. warning:: The MySQL Connector/Python driver published by Oracle is subject 

25 to frequent, major regressions of essential functionality such as being able 

26 to correctly persist simple binary strings which indicate it is not well 

27 tested. The SQLAlchemy project is not able to maintain this dialect fully as 

28 regressions in the driver prevent it from being included in continuous 

29 integration. 

30 

31.. versionchanged:: 2.0.39 

32 

33 The MySQL Connector/Python dialect has been updated to support the 

34 latest version of this DBAPI. Previously, MySQL Connector/Python 

35 was not fully supported. However, support remains limited due to ongoing 

36 regressions introduced in this driver. 

37 

38Connecting to MariaDB with MySQL Connector/Python 

39-------------------------------------------------- 

40 

41MySQL Connector/Python may attempt to pass an incompatible collation to the 

42database when connecting to MariaDB. Experimentation has shown that using 

43``?charset=utf8mb4&collation=utfmb4_general_ci`` or similar MariaDB-compatible 

44charset/collation will allow connectivity. 

45 

46 

47""" # noqa 

48from __future__ import annotations 

49 

50import re 

51from typing import Any 

52from typing import cast 

53from typing import Optional 

54from typing import Sequence 

55from typing import Tuple 

56from typing import TYPE_CHECKING 

57from typing import Union 

58 

59from .base import MariaDBIdentifierPreparer 

60from .base import MySQLCompiler 

61from .base import MySQLDialect 

62from .base import MySQLExecutionContext 

63from .base import MySQLIdentifierPreparer 

64from .mariadb import MariaDBDialect 

65from .types import BIT 

66from ... import util 

67 

68if TYPE_CHECKING: 

69 

70 from ...engine.base import Connection 

71 from ...engine.cursor import CursorResult 

72 from ...engine.interfaces import ConnectArgsType 

73 from ...engine.interfaces import DBAPIConnection 

74 from ...engine.interfaces import DBAPICursor 

75 from ...engine.interfaces import DBAPIModule 

76 from ...engine.interfaces import IsolationLevel 

77 from ...engine.interfaces import PoolProxiedConnection 

78 from ...engine.row import Row 

79 from ...engine.url import URL 

80 from ...sql.elements import BinaryExpression 

81 

82 

83class MySQLExecutionContext_mysqlconnector(MySQLExecutionContext): 

84 def create_server_side_cursor(self) -> DBAPICursor: 

85 return self._dbapi_connection.cursor(buffered=False) 

86 

87 def create_default_cursor(self) -> DBAPICursor: 

88 return self._dbapi_connection.cursor(buffered=True) 

89 

90 

91class MySQLCompiler_mysqlconnector(MySQLCompiler): 

92 def visit_mod_binary( 

93 self, binary: BinaryExpression[Any], operator: Any, **kw: Any 

94 ) -> str: 

95 return ( 

96 self.process(binary.left, **kw) 

97 + " % " 

98 + self.process(binary.right, **kw) 

99 ) 

100 

101 

102class IdentifierPreparerCommon_mysqlconnector: 

103 @property 

104 def _double_percents(self) -> bool: 

105 return False 

106 

107 @_double_percents.setter 

108 def _double_percents(self, value: Any) -> None: 

109 pass 

110 

111 def _escape_identifier(self, value: str) -> str: 

112 value = value.replace( 

113 self.escape_quote, # type:ignore[attr-defined] 

114 self.escape_to_quote, # type:ignore[attr-defined] 

115 ) 

116 return value 

117 

118 

119class MySQLIdentifierPreparer_mysqlconnector( 

120 IdentifierPreparerCommon_mysqlconnector, MySQLIdentifierPreparer 

121): 

122 pass 

123 

124 

125class MariaDBIdentifierPreparer_mysqlconnector( 

126 IdentifierPreparerCommon_mysqlconnector, MariaDBIdentifierPreparer 

127): 

128 pass 

129 

130 

131class _myconnpyBIT(BIT): 

132 def result_processor(self, dialect: Any, coltype: Any) -> None: 

133 """MySQL-connector already converts mysql bits, so.""" 

134 

135 return None 

136 

137 

138class MySQLDialect_mysqlconnector(MySQLDialect): 

139 driver = "mysqlconnector" 

140 supports_statement_cache = True 

141 

142 supports_sane_rowcount = True 

143 supports_sane_multi_rowcount = True 

144 

145 supports_native_decimal = True 

146 

147 supports_native_bit = True 

148 

149 # not until https://bugs.mysql.com/bug.php?id=117548 

150 supports_server_side_cursors = False 

151 

152 default_paramstyle = "format" 

153 statement_compiler = MySQLCompiler_mysqlconnector 

154 

155 execution_ctx_cls = MySQLExecutionContext_mysqlconnector 

156 

157 preparer: type[MySQLIdentifierPreparer] = ( 

158 MySQLIdentifierPreparer_mysqlconnector 

159 ) 

160 

161 colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _myconnpyBIT}) 

162 

163 @classmethod 

164 def import_dbapi(cls) -> DBAPIModule: 

165 return cast("DBAPIModule", __import__("mysql.connector").connector) 

166 

167 def do_ping(self, dbapi_connection: DBAPIConnection) -> bool: 

168 dbapi_connection.ping(False) 

169 return True 

170 

171 def create_connect_args(self, url: URL) -> ConnectArgsType: 

172 opts = url.translate_connect_args(username="user") 

173 

174 opts.update(url.query) 

175 

176 util.coerce_kw_type(opts, "allow_local_infile", bool) 

177 util.coerce_kw_type(opts, "autocommit", bool) 

178 util.coerce_kw_type(opts, "buffered", bool) 

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

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

181 util.coerce_kw_type(opts, "connection_timeout", int) 

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

183 util.coerce_kw_type(opts, "consume_results", bool) 

184 util.coerce_kw_type(opts, "force_ipv6", bool) 

185 util.coerce_kw_type(opts, "get_warnings", bool) 

186 util.coerce_kw_type(opts, "pool_reset_session", bool) 

187 util.coerce_kw_type(opts, "pool_size", int) 

188 util.coerce_kw_type(opts, "raise_on_warnings", bool) 

189 util.coerce_kw_type(opts, "raw", bool) 

190 util.coerce_kw_type(opts, "ssl_verify_cert", bool) 

191 util.coerce_kw_type(opts, "use_pure", bool) 

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

193 

194 # note that "buffered" is set to False by default in MySQL/connector 

195 # python. If you set it to True, then there is no way to get a server 

196 # side cursor because the logic is written to disallow that. 

197 

198 # leaving this at True until 

199 # https://bugs.mysql.com/bug.php?id=117548 can be fixed 

200 opts["buffered"] = True 

201 

202 # FOUND_ROWS must be set in ClientFlag to enable 

203 # supports_sane_rowcount. 

204 if self.dbapi is not None: 

205 try: 

206 from mysql.connector import constants # type: ignore 

207 

208 ClientFlag = constants.ClientFlag 

209 

210 client_flags = opts.get( 

211 "client_flags", ClientFlag.get_default() 

212 ) 

213 client_flags |= ClientFlag.FOUND_ROWS 

214 opts["client_flags"] = client_flags 

215 except Exception: 

216 pass 

217 

218 return [], opts 

219 

220 @util.memoized_property 

221 def _mysqlconnector_version_info(self) -> Optional[Tuple[int, ...]]: 

222 if self.dbapi and hasattr(self.dbapi, "__version__"): 

223 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__) 

224 if m: 

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

226 return None 

227 

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

229 return connection.connection.charset # type: ignore 

230 

231 def _extract_error_code(self, exception: BaseException) -> int: 

232 return exception.errno # type: ignore 

233 

234 def is_disconnect( 

235 self, 

236 e: Exception, 

237 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]], 

238 cursor: Optional[DBAPICursor], 

239 ) -> bool: 

240 errnos = (2006, 2013, 2014, 2045, 2055, 2048) 

241 exceptions = ( 

242 self.loaded_dbapi.OperationalError, # 

243 self.loaded_dbapi.InterfaceError, 

244 self.loaded_dbapi.ProgrammingError, 

245 ) 

246 if isinstance(e, exceptions): 

247 return ( 

248 e.errno in errnos 

249 or "MySQL Connection not available." in str(e) 

250 or "Connection to MySQL is not available" in str(e) 

251 ) 

252 else: 

253 return False 

254 

255 def _compat_fetchall( 

256 self, 

257 rp: CursorResult[Tuple[Any, ...]], 

258 charset: Optional[str] = None, 

259 ) -> Sequence[Row[Tuple[Any, ...]]]: 

260 return rp.fetchall() 

261 

262 def _compat_fetchone( 

263 self, 

264 rp: CursorResult[Tuple[Any, ...]], 

265 charset: Optional[str] = None, 

266 ) -> Optional[Row[Tuple[Any, ...]]]: 

267 return rp.fetchone() 

268 

269 def get_isolation_level_values( 

270 self, dbapi_conn: DBAPIConnection 

271 ) -> Sequence[IsolationLevel]: 

272 return ( 

273 "SERIALIZABLE", 

274 "READ UNCOMMITTED", 

275 "READ COMMITTED", 

276 "REPEATABLE READ", 

277 "AUTOCOMMIT", 

278 ) 

279 

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

281 return bool(dbapi_conn.autocommit) 

282 

283 def set_isolation_level( 

284 self, dbapi_connection: DBAPIConnection, level: IsolationLevel 

285 ) -> None: 

286 if level == "AUTOCOMMIT": 

287 dbapi_connection.autocommit = True 

288 else: 

289 dbapi_connection.autocommit = False 

290 super().set_isolation_level(dbapi_connection, level) 

291 

292 

293class MariaDBDialect_mysqlconnector( 

294 MariaDBDialect, MySQLDialect_mysqlconnector 

295): 

296 supports_statement_cache = True 

297 _allows_uuid_binds = False 

298 preparer = MariaDBIdentifierPreparer_mysqlconnector 

299 

300 

301dialect = MySQLDialect_mysqlconnector 

302mariadb_dialect = MariaDBDialect_mysqlconnector