codekingpro/portable-devtools

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# mypy: ignore-errors
8
9
10r"""
11.. dialect:: mysql+mysqlconnector
12    :name: MySQL Connector/Python
13    :dbapi: myconnpy
14    :connectstring: mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
15    :url: https://pypi.org/project/mysql-connector-python/
16
17.. note::
18
19    The MySQL Connector/Python DBAPI has had many issues since its release,
20    some of which may remain unresolved, and the mysqlconnector dialect is
21    **not tested as part of SQLAlchemy's continuous integration**.
22    The recommended MySQL dialects are mysqlclient and PyMySQL.
23
24"""  # noqa
25
26import re
27
28from .base import BIT
29from .base import MySQLCompiler
30from .base import MySQLDialect
31from .base import MySQLIdentifierPreparer
32from ... import util
33
34
35class MySQLCompiler_mysqlconnector(MySQLCompiler):
36    def visit_mod_binary(self, binary, operator, **kw):
37        return (
38            self.process(binary.left, **kw)
39            + " % "
40            + self.process(binary.right, **kw)
41        )
42
43
44class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
45    @property
46    def _double_percents(self):
47        return False
48
49    @_double_percents.setter
50    def _double_percents(self, value):
51        pass
52
53    def _escape_identifier(self, value):
54        value = value.replace(self.escape_quote, self.escape_to_quote)
55        return value
56
57
58class _myconnpyBIT(BIT):
59    def result_processor(self, dialect, coltype):
60        """MySQL-connector already converts mysql bits, so."""
61
62        return None
63
64
65class MySQLDialect_mysqlconnector(MySQLDialect):
66    driver = "mysqlconnector"
67    supports_statement_cache = True
68
69    supports_sane_rowcount = True
70    supports_sane_multi_rowcount = True
71
72    supports_native_decimal = True
73
74    default_paramstyle = "format"
75    statement_compiler = MySQLCompiler_mysqlconnector
76
77    preparer = MySQLIdentifierPreparer_mysqlconnector
78
79    colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _myconnpyBIT})
80
81    @classmethod
82    def import_dbapi(cls):
83        from mysql import connector
84
85        return connector
86
87    def do_ping(self, dbapi_connection):
88        dbapi_connection.ping(False)
89        return True
90
91    def create_connect_args(self, url):
92        opts = url.translate_connect_args(username="user")
93
94        opts.update(url.query)
95
96        util.coerce_kw_type(opts, "allow_local_infile", bool)
97        util.coerce_kw_type(opts, "autocommit", bool)
98        util.coerce_kw_type(opts, "buffered", bool)
99        util.coerce_kw_type(opts, "compress", bool)
100        util.coerce_kw_type(opts, "connection_timeout", int)
101        util.coerce_kw_type(opts, "connect_timeout", int)
102        util.coerce_kw_type(opts, "consume_results", bool)
103        util.coerce_kw_type(opts, "force_ipv6", bool)
104        util.coerce_kw_type(opts, "get_warnings", bool)
105        util.coerce_kw_type(opts, "pool_reset_session", bool)
106        util.coerce_kw_type(opts, "pool_size", int)
107        util.coerce_kw_type(opts, "raise_on_warnings", bool)
108        util.coerce_kw_type(opts, "raw", bool)
109        util.coerce_kw_type(opts, "ssl_verify_cert", bool)
110        util.coerce_kw_type(opts, "use_pure", bool)
111        util.coerce_kw_type(opts, "use_unicode", bool)
112
113        # unfortunately, MySQL/connector python refuses to release a
114        # cursor without reading fully, so non-buffered isn't an option
115        opts.setdefault("buffered", True)
116
117        # FOUND_ROWS must be set in ClientFlag to enable
118        # supports_sane_rowcount.
119        if self.dbapi is not None:
120            try:
121                from mysql.connector.constants import ClientFlag
122
123                client_flags = opts.get(
124                    "client_flags", ClientFlag.get_default()
125                )
126                client_flags |= ClientFlag.FOUND_ROWS
127                opts["client_flags"] = client_flags
128            except Exception:
129                pass
130        return [[], opts]
131
132    @util.memoized_property
133    def _mysqlconnector_version_info(self):
134        if self.dbapi and hasattr(self.dbapi, "__version__"):
135            m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
136            if m:
137                return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
138
139    def _detect_charset(self, connection):
140        return connection.connection.charset
141
142    def _extract_error_code(self, exception):
143        return exception.errno
144
145    def is_disconnect(self, e, connection, cursor):
146        errnos = (2006, 2013, 2014, 2045, 2055, 2048)
147        exceptions = (self.dbapi.OperationalError, self.dbapi.InterfaceError)
148        if isinstance(e, exceptions):
149            return (
150                e.errno in errnos
151                or "MySQL Connection not available." in str(e)
152                or "Connection to MySQL is not available" in str(e)
153            )
154        else:
155            return False
156
157    def _compat_fetchall(self, rp, charset=None):
158        return rp.fetchall()
159
160    def _compat_fetchone(self, rp, charset=None):
161        return rp.fetchone()
162
163    _isolation_lookup = {
164        "SERIALIZABLE",
165        "READ UNCOMMITTED",
166        "READ COMMITTED",
167        "REPEATABLE READ",
168        "AUTOCOMMIT",
169    }
170
171    def _set_isolation_level(self, connection, level):
172        if level == "AUTOCOMMIT":
173            connection.autocommit = True
174        else:
175            connection.autocommit = False
176            super()._set_isolation_level(connection, level)
177
178
179dialect = MySQLDialect_mysqlconnector
180