codekingpro/portable-devtools

1# dialects/mysql/mariadbconnector.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
10"""
11
12.. dialect:: mysql+mariadbconnector
13    :name: MariaDB Connector/Python
14    :dbapi: mariadb
15    :connectstring: mariadb+mariadbconnector://<user>:<password>@<host>[:<port>]/<dbname>
16    :url: https://pypi.org/project/mariadb/
17
18Driver Status
19-------------
20
21MariaDB Connector/Python enables Python programs to access MariaDB and MySQL
22databases using an API which is compliant with the Python DB API 2.0 (PEP-249).
23It is written in C and uses MariaDB Connector/C client library for client server
24communication.
25
26Note that the default driver for a ``mariadb://`` connection URI continues to
27be ``mysqldb``. ``mariadb+mariadbconnector://`` is required to use this driver.
28
29.. mariadb: https://github.com/mariadb-corporation/mariadb-connector-python
30
31"""  # noqa
32import re
33from uuid import UUID as _python_UUID
34
35from .base import MySQLCompiler
36from .base import MySQLDialect
37from .base import MySQLExecutionContext
38from ... import sql
39from ... import util
40from ...sql import sqltypes
41
42
43mariadb_cpy_minimum_version = (1, 0, 1)
44
45
46class _MariaDBUUID(sqltypes.UUID[sqltypes._UUID_RETURN]):
47    # work around JIRA issue
48    # https://jira.mariadb.org/browse/CONPY-270.  When that issue is fixed,
49    # this type can be removed.
50    def result_processor(self, dialect, coltype):
51        if self.as_uuid:
52
53            def process(value):
54                if value is not None:
55                    if hasattr(value, "decode"):
56                        value = value.decode("ascii")
57                    value = _python_UUID(value)
58                return value
59
60            return process
61        else:
62
63            def process(value):
64                if value is not None:
65                    if hasattr(value, "decode"):
66                        value = value.decode("ascii")
67                    value = str(_python_UUID(value))
68                return value
69
70            return process
71
72
73class MySQLExecutionContext_mariadbconnector(MySQLExecutionContext):
74    _lastrowid = None
75
76    def create_server_side_cursor(self):
77        return self._dbapi_connection.cursor(buffered=False)
78
79    def create_default_cursor(self):
80        return self._dbapi_connection.cursor(buffered=True)
81
82    def post_exec(self):
83        super().post_exec()
84
85        self._rowcount = self.cursor.rowcount
86
87        if self.isinsert and self.compiled.postfetch_lastrowid:
88            self._lastrowid = self.cursor.lastrowid
89
90    def get_lastrowid(self):
91        return self._lastrowid
92
93
94class MySQLCompiler_mariadbconnector(MySQLCompiler):
95    pass
96
97
98class MySQLDialect_mariadbconnector(MySQLDialect):
99    driver = "mariadbconnector"
100    supports_statement_cache = True
101
102    # set this to True at the module level to prevent the driver from running
103    # against a backend that server detects as MySQL. currently this appears to
104    # be unnecessary as MariaDB client libraries have always worked against
105    # MySQL databases.   However, if this changes at some point, this can be
106    # adjusted, but PLEASE ADD A TEST in test/dialect/mysql/test_dialect.py if
107    # this change is made at some point to ensure the correct exception
108    # is raised at the correct point when running the driver against
109    # a MySQL backend.
110    # is_mariadb = True
111
112    supports_unicode_statements = True
113    encoding = "utf8mb4"
114    convert_unicode = True
115    supports_sane_rowcount = True
116    supports_sane_multi_rowcount = True
117    supports_native_decimal = True
118    default_paramstyle = "qmark"
119    execution_ctx_cls = MySQLExecutionContext_mariadbconnector
120    statement_compiler = MySQLCompiler_mariadbconnector
121
122    supports_server_side_cursors = True
123
124    colspecs = util.update_copy(
125        MySQLDialect.colspecs, {sqltypes.Uuid: _MariaDBUUID}
126    )
127
128    @util.memoized_property
129    def _dbapi_version(self):
130        if self.dbapi and hasattr(self.dbapi, "__version__"):
131            return tuple(
132                [
133                    int(x)
134                    for x in re.findall(
135                        r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
136                    )
137                ]
138            )
139        else:
140            return (99, 99, 99)
141
142    def __init__(self, **kwargs):
143        super().__init__(**kwargs)
144        self.paramstyle = "qmark"
145        if self.dbapi is not None:
146            if self._dbapi_version < mariadb_cpy_minimum_version:
147                raise NotImplementedError(
148                    "The minimum required version for MariaDB "
149                    "Connector/Python is %s"
150                    % ".".join(str(x) for x in mariadb_cpy_minimum_version)
151                )
152
153    @classmethod
154    def import_dbapi(cls):
155        return __import__("mariadb")
156
157    def is_disconnect(self, e, connection, cursor):
158        if super().is_disconnect(e, connection, cursor):
159            return True
160        elif isinstance(e, self.dbapi.Error):
161            str_e = str(e).lower()
162            return "not connected" in str_e or "isn't valid" in str_e
163        else:
164            return False
165
166    def create_connect_args(self, url):
167        opts = url.translate_connect_args()
168
169        int_params = [
170            "connect_timeout",
171            "read_timeout",
172            "write_timeout",
173            "client_flag",
174            "port",
175            "pool_size",
176        ]
177        bool_params = [
178            "local_infile",
179            "ssl_verify_cert",
180            "ssl",
181            "pool_reset_connection",
182        ]
183
184        for key in int_params:
185            util.coerce_kw_type(opts, key, int)
186        for key in bool_params:
187            util.coerce_kw_type(opts, key, bool)
188
189        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
190        # supports_sane_rowcount.
191        client_flag = opts.get("client_flag", 0)
192        if self.dbapi is not None:
193            try:
194                CLIENT_FLAGS = __import__(
195                    self.dbapi.__name__ + ".constants.CLIENT"
196                ).constants.CLIENT
197                client_flag |= CLIENT_FLAGS.FOUND_ROWS
198            except (AttributeError, ImportError):
199                self.supports_sane_rowcount = False
200            opts["client_flag"] = client_flag
201        return [[], opts]
202
203    def _extract_error_code(self, exception):
204        try:
205            rc = exception.errno
206        except:
207            rc = -1
208        return rc
209
210    def _detect_charset(self, connection):
211        return "utf8mb4"
212
213    def get_isolation_level_values(self, dbapi_connection):
214        return (
215            "SERIALIZABLE",
216            "READ UNCOMMITTED",
217            "READ COMMITTED",
218            "REPEATABLE READ",
219            "AUTOCOMMIT",
220        )
221
222    def set_isolation_level(self, connection, level):
223        if level == "AUTOCOMMIT":
224            connection.autocommit = True
225        else:
226            connection.autocommit = False
227            super().set_isolation_level(connection, level)
228
229    def do_begin_twophase(self, connection, xid):
230        connection.execute(
231            sql.text("XA BEGIN :xid").bindparams(
232                sql.bindparam("xid", xid, literal_execute=True)
233            )
234        )
235
236    def do_prepare_twophase(self, connection, xid):
237        connection.execute(
238            sql.text("XA END :xid").bindparams(
239                sql.bindparam("xid", xid, literal_execute=True)
240            )
241        )
242        connection.execute(
243            sql.text("XA PREPARE :xid").bindparams(
244                sql.bindparam("xid", xid, literal_execute=True)
245            )
246        )
247
248    def do_rollback_twophase(
249        self, connection, xid, is_prepared=True, recover=False
250    ):
251        if not is_prepared:
252            connection.execute(
253                sql.text("XA END :xid").bindparams(
254                    sql.bindparam("xid", xid, literal_execute=True)
255                )
256            )
257        connection.execute(
258            sql.text("XA ROLLBACK :xid").bindparams(
259                sql.bindparam("xid", xid, literal_execute=True)
260            )
261        )
262
263    def do_commit_twophase(
264        self, connection, xid, is_prepared=True, recover=False
265    ):
266        if not is_prepared:
267            self.do_prepare_twophase(connection, xid)
268        connection.execute(
269            sql.text("XA COMMIT :xid").bindparams(
270                sql.bindparam("xid", xid, literal_execute=True)
271            )
272        )
273
274
275dialect = MySQLDialect_mariadbconnector
276