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

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

8r""" 

9.. dialect:: mysql+mysqlconnector 

10 :name: MySQL Connector/Python 

11 :dbapi: myconnpy 

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

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

14 

15.. note:: 

16 

17 The MySQL Connector/Python DBAPI has had many issues since its release, 

18 some of which may remain unresolved, and the mysqlconnector dialect is 

19 **not tested as part of SQLAlchemy's continuous integration**. 

20 The recommended MySQL dialects are mysqlclient and PyMySQL. 

21 

22""" # noqa 

23 

24import re 

25 

26from .base import BIT 

27from .base import MySQLCompiler 

28from .base import MySQLDialect 

29from .base import MySQLIdentifierPreparer 

30from ... import processors 

31from ... import util 

32 

33 

34class MySQLCompiler_mysqlconnector(MySQLCompiler): 

35 def visit_mod_binary(self, binary, operator, **kw): 

36 if self.dialect._mysqlconnector_double_percents: 

37 return ( 

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

39 + " %% " 

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

41 ) 

42 else: 

43 return ( 

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

45 + " % " 

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

47 ) 

48 

49 def post_process_text(self, text): 

50 if self.dialect._mysqlconnector_double_percents: 

51 return text.replace("%", "%%") 

52 else: 

53 return text 

54 

55 def escape_literal_column(self, text): 

56 if self.dialect._mysqlconnector_double_percents: 

57 return text.replace("%", "%%") 

58 else: 

59 return text 

60 

61 

62class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer): 

63 @property 

64 def _double_percents(self): 

65 return self.dialect._mysqlconnector_double_percents 

66 

67 @_double_percents.setter 

68 def _double_percents(self, value): 

69 pass 

70 

71 def _escape_identifier(self, value): 

72 value = value.replace(self.escape_quote, self.escape_to_quote) 

73 if self.dialect._mysqlconnector_double_percents: 

74 return value.replace("%", "%%") 

75 else: 

76 return value 

77 

78 

79class _myconnpyBIT(BIT): 

80 def result_processor(self, dialect, coltype): 

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

82 

83 return None 

84 

85 

86class MySQLDialect_mysqlconnector(MySQLDialect): 

87 driver = "mysqlconnector" 

88 supports_statement_cache = True 

89 

90 supports_unicode_binds = True 

91 

92 supports_sane_rowcount = True 

93 supports_sane_multi_rowcount = True 

94 

95 supports_native_decimal = True 

96 

97 default_paramstyle = "format" 

98 statement_compiler = MySQLCompiler_mysqlconnector 

99 

100 preparer = MySQLIdentifierPreparer_mysqlconnector 

101 

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

103 

104 def __init__(self, *arg, **kw): 

105 super(MySQLDialect_mysqlconnector, self).__init__(*arg, **kw) 

106 

107 # hack description encoding since mysqlconnector randomly 

108 # returns bytes or not 

109 self._description_decoder = ( 

110 processors.to_conditional_unicode_processor_factory 

111 )(self.description_encoding) 

112 

113 def _check_unicode_description(self, connection): 

114 # hack description encoding since mysqlconnector randomly 

115 # returns bytes or not 

116 return False 

117 

118 @property 

119 def description_encoding(self): 

120 # total guess 

121 return "latin-1" 

122 

123 @util.memoized_property 

124 def supports_unicode_statements(self): 

125 return util.py3k or self._mysqlconnector_version_info > (2, 0) 

126 

127 @classmethod 

128 def dbapi(cls): 

129 from mysql import connector 

130 

131 return connector 

132 

133 def do_ping(self, dbapi_connection): 

134 try: 

135 dbapi_connection.ping(False) 

136 except self.dbapi.Error as err: 

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

138 return False 

139 else: 

140 raise 

141 else: 

142 return True 

143 

144 def create_connect_args(self, url): 

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

146 

147 opts.update(url.query) 

148 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

165 

166 # unfortunately, MySQL/connector python refuses to release a 

167 # cursor without reading fully, so non-buffered isn't an option 

168 opts.setdefault("buffered", True) 

169 

170 # FOUND_ROWS must be set in ClientFlag to enable 

171 # supports_sane_rowcount. 

172 if self.dbapi is not None: 

173 try: 

174 from mysql.connector.constants import ClientFlag 

175 

176 client_flags = opts.get( 

177 "client_flags", ClientFlag.get_default() 

178 ) 

179 client_flags |= ClientFlag.FOUND_ROWS 

180 opts["client_flags"] = client_flags 

181 except Exception: 

182 pass 

183 return [[], opts] 

184 

185 @util.memoized_property 

186 def _mysqlconnector_version_info(self): 

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

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

189 if m: 

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

191 

192 @util.memoized_property 

193 def _mysqlconnector_double_percents(self): 

194 return not util.py3k and self._mysqlconnector_version_info < (2, 0) 

195 

196 def _detect_charset(self, connection): 

197 return connection.connection.charset 

198 

199 def _extract_error_code(self, exception): 

200 return exception.errno 

201 

202 def is_disconnect(self, e, connection, cursor): 

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

204 exceptions = (self.dbapi.OperationalError, self.dbapi.InterfaceError) 

205 if isinstance(e, exceptions): 

206 return ( 

207 e.errno in errnos 

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

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

210 ) 

211 else: 

212 return False 

213 

214 def _compat_fetchall(self, rp, charset=None): 

215 return rp.fetchall() 

216 

217 def _compat_fetchone(self, rp, charset=None): 

218 return rp.fetchone() 

219 

220 _isolation_lookup = set( 

221 [ 

222 "SERIALIZABLE", 

223 "READ UNCOMMITTED", 

224 "READ COMMITTED", 

225 "REPEATABLE READ", 

226 "AUTOCOMMIT", 

227 ] 

228 ) 

229 

230 def _set_isolation_level(self, connection, level): 

231 if level == "AUTOCOMMIT": 

232 connection.autocommit = True 

233 else: 

234 connection.autocommit = False 

235 super(MySQLDialect_mysqlconnector, self)._set_isolation_level( 

236 connection, level 

237 ) 

238 

239 

240dialect = MySQLDialect_mysqlconnector