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

1# mysql/oursql.py 

2# Copyright (C) 2005-2023 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+oursql 

11 :name: OurSQL 

12 :dbapi: oursql 

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

14 :url: https://packages.python.org/oursql/ 

15 

16.. note:: 

17 

18 The OurSQL MySQL dialect is legacy and is no longer supported upstream, 

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

20 The recommended MySQL dialects are mysqlclient and PyMySQL. 

21 

22.. deprecated:: 1.4 The OurSQL DBAPI is deprecated and will be removed 

23 in a future version. Please use one of the supported DBAPIs to 

24 connect to mysql. 

25 

26Unicode 

27------- 

28 

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

30handling. 

31 

32 

33""" 

34 

35 

36from .base import BIT 

37from .base import MySQLDialect 

38from .base import MySQLExecutionContext 

39from ... import types as sqltypes 

40from ... import util 

41 

42 

43class _oursqlBIT(BIT): 

44 def result_processor(self, dialect, coltype): 

45 """oursql already converts mysql bits, so.""" 

46 

47 return None 

48 

49 

50class MySQLExecutionContext_oursql(MySQLExecutionContext): 

51 @property 

52 def plain_query(self): 

53 return self.execution_options.get("_oursql_plain_query", False) 

54 

55 

56class MySQLDialect_oursql(MySQLDialect): 

57 driver = "oursql" 

58 supports_statement_cache = True 

59 

60 if util.py2k: 

61 supports_unicode_binds = True 

62 supports_unicode_statements = True 

63 

64 supports_native_decimal = True 

65 

66 supports_sane_rowcount = True 

67 supports_sane_multi_rowcount = True 

68 execution_ctx_cls = MySQLExecutionContext_oursql 

69 

70 colspecs = util.update_copy( 

71 MySQLDialect.colspecs, {sqltypes.Time: sqltypes.Time, BIT: _oursqlBIT} 

72 ) 

73 

74 @classmethod 

75 def dbapi(cls): 

76 util.warn_deprecated( 

77 "The OurSQL DBAPI is deprecated and will be removed " 

78 "in a future version. Please use one of the supported DBAPIs to " 

79 "connect to mysql.", 

80 version="1.4", 

81 ) 

82 return __import__("oursql") 

83 

84 def do_execute(self, cursor, statement, parameters, context=None): 

85 """Provide an implementation of 

86 *cursor.execute(statement, parameters)*.""" 

87 

88 if context and context.plain_query: 

89 cursor.execute(statement, plain_query=True) 

90 else: 

91 cursor.execute(statement, parameters) 

92 

93 def do_begin(self, connection): 

94 connection.cursor().execute("BEGIN", plain_query=True) 

95 

96 def _xa_query(self, connection, query, xid): 

97 if util.py2k: 

98 arg = connection.connection._escape_string(xid) 

99 else: 

100 charset = self._connection_charset 

101 arg = connection.connection._escape_string( 

102 xid.encode(charset) 

103 ).decode(charset) 

104 arg = "'%s'" % arg 

105 connection.execution_options(_oursql_plain_query=True).exec_driver_sql( 

106 query % arg 

107 ) 

108 

109 # Because mysql is bad, these methods have to be 

110 # reimplemented to use _PlainQuery. Basically, some queries 

111 # refuse to return any data if they're run through 

112 # the parameterized query API, or refuse to be parameterized 

113 # in the first place. 

114 def do_begin_twophase(self, connection, xid): 

115 self._xa_query(connection, "XA BEGIN %s", xid) 

116 

117 def do_prepare_twophase(self, connection, xid): 

118 self._xa_query(connection, "XA END %s", xid) 

119 self._xa_query(connection, "XA PREPARE %s", xid) 

120 

121 def do_rollback_twophase( 

122 self, connection, xid, is_prepared=True, recover=False 

123 ): 

124 if not is_prepared: 

125 self._xa_query(connection, "XA END %s", xid) 

126 self._xa_query(connection, "XA ROLLBACK %s", xid) 

127 

128 def do_commit_twophase( 

129 self, connection, xid, is_prepared=True, recover=False 

130 ): 

131 if not is_prepared: 

132 self.do_prepare_twophase(connection, xid) 

133 self._xa_query(connection, "XA COMMIT %s", xid) 

134 

135 # Q: why didn't we need all these "plain_query" overrides earlier ? 

136 # am i on a newer/older version of OurSQL ? 

137 def has_table(self, connection, table_name, schema=None): 

138 return MySQLDialect.has_table( 

139 self, 

140 connection.connect().execution_options(_oursql_plain_query=True), 

141 table_name, 

142 schema, 

143 ) 

144 

145 def get_table_options(self, connection, table_name, schema=None, **kw): 

146 return MySQLDialect.get_table_options( 

147 self, 

148 connection.connect().execution_options(_oursql_plain_query=True), 

149 table_name, 

150 schema=schema, 

151 **kw 

152 ) 

153 

154 def get_columns(self, connection, table_name, schema=None, **kw): 

155 return MySQLDialect.get_columns( 

156 self, 

157 connection.connect().execution_options(_oursql_plain_query=True), 

158 table_name, 

159 schema=schema, 

160 **kw 

161 ) 

162 

163 def get_view_names(self, connection, schema=None, **kw): 

164 return MySQLDialect.get_view_names( 

165 self, 

166 connection.connect().execution_options(_oursql_plain_query=True), 

167 schema=schema, 

168 **kw 

169 ) 

170 

171 def get_table_names(self, connection, schema=None, **kw): 

172 return MySQLDialect.get_table_names( 

173 self, 

174 connection.connect().execution_options(_oursql_plain_query=True), 

175 schema, 

176 ) 

177 

178 def get_schema_names(self, connection, **kw): 

179 return MySQLDialect.get_schema_names( 

180 self, 

181 connection.connect().execution_options(_oursql_plain_query=True), 

182 **kw 

183 ) 

184 

185 def initialize(self, connection): 

186 return MySQLDialect.initialize( 

187 self, connection.execution_options(_oursql_plain_query=True) 

188 ) 

189 

190 def _show_create_table( 

191 self, connection, table, charset=None, full_name=None 

192 ): 

193 return MySQLDialect._show_create_table( 

194 self, 

195 connection.connect(close_with_result=True).execution_options( 

196 _oursql_plain_query=True 

197 ), 

198 table, 

199 charset, 

200 full_name, 

201 ) 

202 

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

204 if isinstance(e, self.dbapi.ProgrammingError): 

205 return ( 

206 e.errno is None 

207 and "cursor" not in e.args[1] 

208 and e.args[1].endswith("closed") 

209 ) 

210 else: 

211 return e.errno in (2006, 2013, 2014, 2045, 2055) 

212 

213 def create_connect_args(self, url): 

214 opts = url.translate_connect_args( 

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

216 ) 

217 opts.update(url.query) 

218 

219 util.coerce_kw_type(opts, "port", int) 

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

221 util.coerce_kw_type(opts, "autoping", bool) 

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

223 

224 util.coerce_kw_type(opts, "default_charset", bool) 

225 if opts.pop("default_charset", False): 

226 opts["charset"] = None 

227 else: 

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

229 opts["use_unicode"] = opts.get("use_unicode", True) 

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

231 

232 # FOUND_ROWS must be set in CLIENT_FLAGS to enable 

233 # supports_sane_rowcount. 

234 opts.setdefault("found_rows", True) 

235 

236 ssl = {} 

237 for key in [ 

238 "ssl_ca", 

239 "ssl_key", 

240 "ssl_cert", 

241 "ssl_capath", 

242 "ssl_cipher", 

243 ]: 

244 if key in opts: 

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

246 util.coerce_kw_type(ssl, key[4:], str) 

247 del opts[key] 

248 if ssl: 

249 opts["ssl"] = ssl 

250 

251 return [[], opts] 

252 

253 def _extract_error_code(self, exception): 

254 return exception.errno 

255 

256 def _detect_charset(self, connection): 

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

258 

259 return connection.connection.charset 

260 

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

262 """oursql isn't super-broken like MySQLdb, yaaay.""" 

263 return rp.fetchall() 

264 

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

266 """oursql isn't super-broken like MySQLdb, yaaay.""" 

267 return rp.fetchone() 

268 

269 def _compat_first(self, rp, charset=None): 

270 return rp.first() 

271 

272 

273dialect = MySQLDialect_oursql