1# mysql/oursql.py
2# Copyright (C) 2005-2021 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
274