codekingpro/portable-devtools

1# dialects/mysql/mysqldb.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+mysqldb
13    :name: mysqlclient (maintained fork of MySQL-Python)
14    :dbapi: mysqldb
15    :connectstring: mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
16    :url: https://pypi.org/project/mysqlclient/
17
18Driver Status
19-------------
20
21The mysqlclient DBAPI is a maintained fork of the
22`MySQL-Python <https://sourceforge.net/projects/mysql-python>`_ DBAPI
23that is no longer maintained.  `mysqlclient`_ supports Python 2 and Python 3
24and is very stable.
25
26.. _mysqlclient: https://github.com/PyMySQL/mysqlclient-python
27
28.. _mysqldb_unicode:
29
30Unicode
31-------
32
33Please see :ref:`mysql_unicode` for current recommendations on unicode
34handling.
35
36.. _mysqldb_ssl:
37
38SSL Connections
39----------------
40
41The mysqlclient and PyMySQL DBAPIs accept an additional dictionary under the
42key "ssl", which may be specified using the
43:paramref:`_sa.create_engine.connect_args` dictionary::
44
45    engine = create_engine(
46        "mysql+mysqldb://scott:tiger@192.168.0.134/test",
47        connect_args={
48            "ssl": {
49                "ca": "/home/gord/client-ssl/ca.pem",
50                "cert": "/home/gord/client-ssl/client-cert.pem",
51                "key": "/home/gord/client-ssl/client-key.pem"
52            }
53        }
54    )
55
56For convenience, the following keys may also be specified inline within the URL
57where they will be interpreted into the "ssl" dictionary automatically:
58"ssl_ca", "ssl_cert", "ssl_key", "ssl_capath", "ssl_cipher",
59"ssl_check_hostname". An example is as follows::
60
61    connection_uri = (
62        "mysql+mysqldb://scott:tiger@192.168.0.134/test"
63        "?ssl_ca=/home/gord/client-ssl/ca.pem"
64        "&ssl_cert=/home/gord/client-ssl/client-cert.pem"
65        "&ssl_key=/home/gord/client-ssl/client-key.pem"
66    )
67
68.. seealso::
69
70    :ref:`pymysql_ssl` in the PyMySQL dialect
71
72
73Using MySQLdb with Google Cloud SQL
74-----------------------------------
75
76Google Cloud SQL now recommends use of the MySQLdb dialect.  Connect
77using a URL like the following::
78
79    mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/<projectid>:<instancename>
80
81Server Side Cursors
82-------------------
83
84The mysqldb dialect supports server-side cursors. See :ref:`mysql_ss_cursors`.
85
86"""
87
88import re
89
90from .base import MySQLCompiler
91from .base import MySQLDialect
92from .base import MySQLExecutionContext
93from .base import MySQLIdentifierPreparer
94from .base import TEXT
95from ... import sql
96from ... import util
97
98
99class MySQLExecutionContext_mysqldb(MySQLExecutionContext):
100    pass
101
102
103class MySQLCompiler_mysqldb(MySQLCompiler):
104    pass
105
106
107class MySQLDialect_mysqldb(MySQLDialect):
108    driver = "mysqldb"
109    supports_statement_cache = True
110    supports_unicode_statements = True
111    supports_sane_rowcount = True
112    supports_sane_multi_rowcount = True
113
114    supports_native_decimal = True
115
116    default_paramstyle = "format"
117    execution_ctx_cls = MySQLExecutionContext_mysqldb
118    statement_compiler = MySQLCompiler_mysqldb
119    preparer = MySQLIdentifierPreparer
120
121    def __init__(self, **kwargs):
122        super().__init__(**kwargs)
123        self._mysql_dbapi_version = (
124            self._parse_dbapi_version(self.dbapi.__version__)
125            if self.dbapi is not None and hasattr(self.dbapi, "__version__")
126            else (0, 0, 0)
127        )
128
129    def _parse_dbapi_version(self, version):
130        m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)
131        if m:
132            return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
133        else:
134            return (0, 0, 0)
135
136    @util.langhelpers.memoized_property
137    def supports_server_side_cursors(self):
138        try:
139            cursors = __import__("MySQLdb.cursors").cursors
140            self._sscursor = cursors.SSCursor
141            return True
142        except (ImportError, AttributeError):
143            return False
144
145    @classmethod
146    def import_dbapi(cls):
147        return __import__("MySQLdb")
148
149    def on_connect(self):
150        super_ = super().on_connect()
151
152        def on_connect(conn):
153            if super_ is not None:
154                super_(conn)
155
156            charset_name = conn.character_set_name()
157
158            if charset_name is not None:
159                cursor = conn.cursor()
160                cursor.execute("SET NAMES %s" % charset_name)
161                cursor.close()
162
163        return on_connect
164
165    def do_ping(self, dbapi_connection):
166        dbapi_connection.ping()
167        return True
168
169    def do_executemany(self, cursor, statement, parameters, context=None):
170        rowcount = cursor.executemany(statement, parameters)
171        if context is not None:
172            context._rowcount = rowcount
173
174    def _check_unicode_returns(self, connection):
175        # work around issue fixed in
176        # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8
177        # specific issue w/ the utf8mb4_bin collation and unicode returns
178
179        collation = connection.exec_driver_sql(
180            "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"
181            % (
182                self.identifier_preparer.quote("Charset"),
183                self.identifier_preparer.quote("Collation"),
184            )
185        ).scalar()
186        has_utf8mb4_bin = self.server_version_info > (5,) and collation
187        if has_utf8mb4_bin:
188            additional_tests = [
189                sql.collate(
190                    sql.cast(
191                        sql.literal_column("'test collated returns'"),
192                        TEXT(charset="utf8mb4"),
193                    ),
194                    "utf8mb4_bin",
195                )
196            ]
197        else:
198            additional_tests = []
199        return super()._check_unicode_returns(connection, additional_tests)
200
201    def create_connect_args(self, url, _translate_args=None):
202        if _translate_args is None:
203            _translate_args = dict(
204                database="db", username="user", password="passwd"
205            )
206
207        opts = url.translate_connect_args(**_translate_args)
208        opts.update(url.query)
209
210        util.coerce_kw_type(opts, "compress", bool)
211        util.coerce_kw_type(opts, "connect_timeout", int)
212        util.coerce_kw_type(opts, "read_timeout", int)
213        util.coerce_kw_type(opts, "write_timeout", int)
214        util.coerce_kw_type(opts, "client_flag", int)
215        util.coerce_kw_type(opts, "local_infile", int)
216        # Note: using either of the below will cause all strings to be
217        # returned as Unicode, both in raw SQL operations and with column
218        # types like String and MSString.
219        util.coerce_kw_type(opts, "use_unicode", bool)
220        util.coerce_kw_type(opts, "charset", str)
221
222        # Rich values 'cursorclass' and 'conv' are not supported via
223        # query string.
224
225        ssl = {}
226        keys = [
227            ("ssl_ca", str),
228            ("ssl_key", str),
229            ("ssl_cert", str),
230            ("ssl_capath", str),
231            ("ssl_cipher", str),
232            ("ssl_check_hostname", bool),
233        ]
234        for key, kw_type in keys:
235            if key in opts:
236                ssl[key[4:]] = opts[key]
237                util.coerce_kw_type(ssl, key[4:], kw_type)
238                del opts[key]
239        if ssl:
240            opts["ssl"] = ssl
241
242        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
243        # supports_sane_rowcount.
244        client_flag = opts.get("client_flag", 0)
245
246        client_flag_found_rows = self._found_rows_client_flag()
247        if client_flag_found_rows is not None:
248            client_flag |= client_flag_found_rows
249            opts["client_flag"] = client_flag
250        return [[], opts]
251
252    def _found_rows_client_flag(self):
253        if self.dbapi is not None:
254            try:
255                CLIENT_FLAGS = __import__(
256                    self.dbapi.__name__ + ".constants.CLIENT"
257                ).constants.CLIENT
258            except (AttributeError, ImportError):
259                return None
260            else:
261                return CLIENT_FLAGS.FOUND_ROWS
262        else:
263            return None
264
265    def _extract_error_code(self, exception):
266        return exception.args[0]
267
268    def _detect_charset(self, connection):
269        """Sniff out the character set in use for connection results."""
270
271        try:
272            # note: the SQL here would be
273            # "SHOW VARIABLES LIKE 'character_set%%'"
274            cset_name = connection.connection.character_set_name
275        except AttributeError:
276            util.warn(
277                "No 'character_set_name' can be detected with "
278                "this MySQL-Python version; "
279                "please upgrade to a recent version of MySQL-Python.  "
280                "Assuming latin1."
281            )
282            return "latin1"
283        else:
284            return cset_name()
285
286    def get_isolation_level_values(self, dbapi_connection):
287        return (
288            "SERIALIZABLE",
289            "READ UNCOMMITTED",
290            "READ COMMITTED",
291            "REPEATABLE READ",
292            "AUTOCOMMIT",
293        )
294
295    def set_isolation_level(self, dbapi_connection, level):
296        if level == "AUTOCOMMIT":
297            dbapi_connection.autocommit(True)
298        else:
299            dbapi_connection.autocommit(False)
300            super().set_isolation_level(dbapi_connection, level)
301
302
303dialect = MySQLDialect_mysqldb
304