Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# dialects/mysql/oursql.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
8"""
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/
16.. note::
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.
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.
26Unicode
27-------
29Please see :ref:`mysql_unicode` for current recommendations on unicode
30handling.
33"""
36from .base import BIT
37from .base import MySQLDialect
38from .base import MySQLExecutionContext
39from ... import types as sqltypes
40from ... import util
43class _oursqlBIT(BIT):
44 def result_processor(self, dialect, coltype):
45 """oursql already converts mysql bits, so."""
47 return None
50class MySQLExecutionContext_oursql(MySQLExecutionContext):
51 @property
52 def plain_query(self):
53 return self.execution_options.get("_oursql_plain_query", False)
56class MySQLDialect_oursql(MySQLDialect):
57 driver = "oursql"
58 supports_statement_cache = True
60 if util.py2k:
61 supports_unicode_binds = True
62 supports_unicode_statements = True
64 supports_native_decimal = True
66 supports_sane_rowcount = True
67 supports_sane_multi_rowcount = True
68 execution_ctx_cls = MySQLExecutionContext_oursql
70 colspecs = util.update_copy(
71 MySQLDialect.colspecs, {sqltypes.Time: sqltypes.Time, BIT: _oursqlBIT}
72 )
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")
84 def do_execute(self, cursor, statement, parameters, context=None):
85 """Provide an implementation of
86 *cursor.execute(statement, parameters)*."""
88 if context and context.plain_query:
89 cursor.execute(statement, plain_query=True)
90 else:
91 cursor.execute(statement, parameters)
93 def do_begin(self, connection):
94 connection.cursor().execute("BEGIN", plain_query=True)
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 )
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)
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)
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)
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)
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 )
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 )
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 )
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 )
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 )
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 )
185 def initialize(self, connection):
186 return MySQLDialect.initialize(
187 self, connection.execution_options(_oursql_plain_query=True)
188 )
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 )
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)
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)
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)
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)
232 # FOUND_ROWS must be set in CLIENT_FLAGS to enable
233 # supports_sane_rowcount.
234 opts.setdefault("found_rows", True)
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
251 return [[], opts]
253 def _extract_error_code(self, exception):
254 return exception.errno
256 def _detect_charset(self, connection):
257 """Sniff out the character set in use for connection results."""
259 return connection.connection.charset
261 def _compat_fetchall(self, rp, charset=None):
262 """oursql isn't super-broken like MySQLdb, yaaay."""
263 return rp.fetchall()
265 def _compat_fetchone(self, rp, charset=None):
266 """oursql isn't super-broken like MySQLdb, yaaay."""
267 return rp.fetchone()
269 def _compat_first(self, rp, charset=None):
270 return rp.first()
273dialect = MySQLDialect_oursql