codekingpro/portable-devtools

1# dialects/mysql/mysqlconnector.py
2# Copyright (C) 2005-2026 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
9r"""
10.. dialect:: mysql+mysqlconnector
11    :name: MySQL Connector/Python
12    :dbapi: myconnpy
13    :connectstring: mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
14    :url: https://pypi.org/project/mysql-connector-python/
15
16Driver Status
17-------------
18
19MySQL Connector/Python is supported as of SQLAlchemy 2.0.39 to the
20degree which the driver is functional.   There are still ongoing issues
21with features such as server side cursors which remain disabled until
22upstream issues are repaired.
23
24.. warning:: The MySQL Connector/Python driver published by Oracle is subject
25   to frequent, major regressions of essential functionality such as being able
26   to correctly persist simple binary strings which indicate it is not well
27   tested.  The SQLAlchemy project is not able to maintain this dialect fully as
28   regressions in the driver prevent it from being included in continuous
29   integration.
30
31.. versionchanged:: 2.0.39
32
33    The MySQL Connector/Python dialect has been updated to support the
34    latest version of this DBAPI.   Previously, MySQL Connector/Python
35    was not fully supported.  However, support remains limited due to ongoing
36    regressions introduced in this driver.
37
38Connecting to MariaDB with MySQL Connector/Python
39--------------------------------------------------
40
41MySQL Connector/Python may attempt to pass an incompatible collation to the
42database when connecting to MariaDB.  Experimentation has shown that using
43``?charset=utf8mb4&collation=utfmb4_general_ci`` or similar MariaDB-compatible
44charset/collation will allow connectivity.
45
46
47"""  # noqa
48from __future__ import annotations
49
50import re
51from typing import Any
52from typing import cast
53from typing import Optional
54from typing import Sequence
55from typing import Tuple
56from typing import TYPE_CHECKING
57from typing import Union
58
59from .base import MariaDBIdentifierPreparer
60from .base import MySQLCompiler
61from .base import MySQLDialect
62from .base import MySQLExecutionContext
63from .base import MySQLIdentifierPreparer
64from .mariadb import MariaDBDialect
65from .types import BIT
66from ... import util
67
68if TYPE_CHECKING:
69
70    from ...engine.base import Connection
71    from ...engine.cursor import CursorResult
72    from ...engine.interfaces import ConnectArgsType
73    from ...engine.interfaces import DBAPIConnection
74    from ...engine.interfaces import DBAPICursor
75    from ...engine.interfaces import DBAPIModule
76    from ...engine.interfaces import IsolationLevel
77    from ...engine.interfaces import PoolProxiedConnection
78    from ...engine.row import Row
79    from ...engine.url import URL
80    from ...sql.elements import BinaryExpression
81
82
83class MySQLExecutionContext_mysqlconnector(MySQLExecutionContext):
84    def create_server_side_cursor(self) -> DBAPICursor:
85        return self._dbapi_connection.cursor(buffered=False)
86
87    def create_default_cursor(self) -> DBAPICursor:
88        return self._dbapi_connection.cursor(buffered=True)
89
90
91class MySQLCompiler_mysqlconnector(MySQLCompiler):
92    def visit_mod_binary(
93        self, binary: BinaryExpression[Any], operator: Any, **kw: Any
94    ) -> str:
95        return (
96            self.process(binary.left, **kw)
97            + " % "
98            + self.process(binary.right, **kw)
99        )
100
101
102class IdentifierPreparerCommon_mysqlconnector:
103    @property
104    def _double_percents(self) -> bool:
105        return False
106
107    @_double_percents.setter
108    def _double_percents(self, value: Any) -> None:
109        pass
110
111    def _escape_identifier(self, value: str) -> str:
112        value = value.replace(
113            self.escape_quote,  # type:ignore[attr-defined]
114            self.escape_to_quote,  # type:ignore[attr-defined]
115        )
116        return value
117
118
119class MySQLIdentifierPreparer_mysqlconnector(
120    IdentifierPreparerCommon_mysqlconnector, MySQLIdentifierPreparer
121):
122    pass
123
124
125class MariaDBIdentifierPreparer_mysqlconnector(
126    IdentifierPreparerCommon_mysqlconnector, MariaDBIdentifierPreparer
127):
128    pass
129
130
131class _myconnpyBIT(BIT):
132    def result_processor(self, dialect: Any, coltype: Any) -> None:
133        """MySQL-connector already converts mysql bits, so."""
134
135        return None
136
137
138class MySQLDialect_mysqlconnector(MySQLDialect):
139    driver = "mysqlconnector"
140    supports_statement_cache = True
141
142    supports_sane_rowcount = True
143    supports_sane_multi_rowcount = True
144
145    supports_native_decimal = True
146
147    supports_native_bit = True
148
149    # not until https://bugs.mysql.com/bug.php?id=117548
150    supports_server_side_cursors = False
151
152    default_paramstyle = "format"
153    statement_compiler = MySQLCompiler_mysqlconnector
154
155    execution_ctx_cls = MySQLExecutionContext_mysqlconnector
156
157    preparer: type[MySQLIdentifierPreparer] = (
158        MySQLIdentifierPreparer_mysqlconnector
159    )
160
161    colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _myconnpyBIT})
162
163    @classmethod
164    def import_dbapi(cls) -> DBAPIModule:
165        return cast("DBAPIModule", __import__("mysql.connector").connector)
166
167    def do_ping(self, dbapi_connection: DBAPIConnection) -> bool:
168        dbapi_connection.ping(False)
169        return True
170
171    def create_connect_args(self, url: URL) -> ConnectArgsType:
172        opts = url.translate_connect_args(username="user")
173
174        opts.update(url.query)
175
176        util.coerce_kw_type(opts, "allow_local_infile", bool)
177        util.coerce_kw_type(opts, "autocommit", bool)
178        util.coerce_kw_type(opts, "buffered", bool)
179        util.coerce_kw_type(opts, "client_flag", int)
180        util.coerce_kw_type(opts, "compress", bool)
181        util.coerce_kw_type(opts, "connection_timeout", int)
182        util.coerce_kw_type(opts, "connect_timeout", int)
183        util.coerce_kw_type(opts, "consume_results", bool)
184        util.coerce_kw_type(opts, "force_ipv6", bool)
185        util.coerce_kw_type(opts, "get_warnings", bool)
186        util.coerce_kw_type(opts, "pool_reset_session", bool)
187        util.coerce_kw_type(opts, "pool_size", int)
188        util.coerce_kw_type(opts, "raise_on_warnings", bool)
189        util.coerce_kw_type(opts, "raw", bool)
190        util.coerce_kw_type(opts, "ssl_verify_cert", bool)
191        util.coerce_kw_type(opts, "use_pure", bool)
192        util.coerce_kw_type(opts, "use_unicode", bool)
193
194        # note that "buffered" is set to False by default in MySQL/connector
195        # python.  If you set it to True, then there is no way to get a server
196        # side cursor because the logic is written to disallow that.
197
198        # leaving this at True until
199        # https://bugs.mysql.com/bug.php?id=117548 can be fixed
200        opts["buffered"] = True
201
202        # FOUND_ROWS must be set in ClientFlag to enable
203        # supports_sane_rowcount.
204        if self.dbapi is not None:
205            try:
206                from mysql.connector import constants  # type: ignore
207
208                ClientFlag = constants.ClientFlag
209
210                client_flags = opts.get(
211                    "client_flags", ClientFlag.get_default()
212                )
213                client_flags |= ClientFlag.FOUND_ROWS
214                opts["client_flags"] = client_flags
215            except Exception:
216                pass
217
218        return [], opts
219
220    @util.memoized_property
221    def _mysqlconnector_version_info(self) -> Optional[Tuple[int, ...]]:
222        if self.dbapi and hasattr(self.dbapi, "__version__"):
223            m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
224            if m:
225                return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
226        return None
227
228    def _detect_charset(self, connection: Connection) -> str:
229        return connection.connection.charset  # type: ignore
230
231    def _extract_error_code(self, exception: BaseException) -> int:
232        return exception.errno  # type: ignore
233
234    def is_disconnect(
235        self,
236        e: Exception,
237        connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
238        cursor: Optional[DBAPICursor],
239    ) -> bool:
240        errnos = (2006, 2013, 2014, 2045, 2055, 2048)
241        exceptions = (
242            self.loaded_dbapi.OperationalError,  #
243            self.loaded_dbapi.InterfaceError,
244            self.loaded_dbapi.ProgrammingError,
245        )
246        if isinstance(e, exceptions):
247            return (
248                e.errno in errnos
249                or "MySQL Connection not available." in str(e)
250                or "Connection to MySQL is not available" in str(e)
251            )
252        else:
253            return False
254
255    def _compat_fetchall(
256        self,
257        rp: CursorResult[Tuple[Any, ...]],
258        charset: Optional[str] = None,
259    ) -> Sequence[Row[Tuple[Any, ...]]]:
260        return rp.fetchall()
261
262    def _compat_fetchone(
263        self,
264        rp: CursorResult[Tuple[Any, ...]],
265        charset: Optional[str] = None,
266    ) -> Optional[Row[Tuple[Any, ...]]]:
267        return rp.fetchone()
268
269    def get_isolation_level_values(
270        self, dbapi_conn: DBAPIConnection
271    ) -> Sequence[IsolationLevel]:
272        return (
273            "SERIALIZABLE",
274            "READ UNCOMMITTED",
275            "READ COMMITTED",
276            "REPEATABLE READ",
277            "AUTOCOMMIT",
278        )
279
280    def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
281        return bool(dbapi_conn.autocommit)
282
283    def set_isolation_level(
284        self, dbapi_connection: DBAPIConnection, level: IsolationLevel
285    ) -> None:
286        if level == "AUTOCOMMIT":
287            dbapi_connection.autocommit = True
288        else:
289            dbapi_connection.autocommit = False
290            super().set_isolation_level(dbapi_connection, level)
291
292
293class MariaDBDialect_mysqlconnector(
294    MariaDBDialect, MySQLDialect_mysqlconnector
295):
296    supports_statement_cache = True
297    _allows_uuid_binds = False
298    preparer = MariaDBIdentifierPreparer_mysqlconnector
299
300
301dialect = MySQLDialect_mysqlconnector
302mariadb_dialect = MariaDBDialect_mysqlconnector
303