✘✘ GRAYBYTE WORDPRESS FILE MANAGER ✘✘

# dialects/mysql/mysqldb.py
# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors


"""

.. dialect:: mysql+mysqldb
    :name: mysqlclient (maintained fork of MySQL-Python)
    :dbapi: mysqldb
    :connectstring: mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
    :url: https://pypi.org/project/mysqlclient/

Driver Status
-------------

The mysqlclient DBAPI is a maintained fork of the
`MySQL-Python <https://sourceforge.net/projects/mysql-python>`_ DBAPI
that is no longer maintained.  `mysqlclient`_ supports Python 2 and Python 3
and is very stable.

.. _mysqlclient: https://github.com/PyMySQL/mysqlclient-python

.. _mysqldb_unicode:

Unicode
-------

Please see :ref:`mysql_unicode` for current recommendations on unicode
handling.

.. _mysqldb_ssl:

SSL Connections
----------------

The mysqlclient and PyMySQL DBAPIs accept an additional dictionary under the
key "ssl", which may be specified using the
:paramref:`_sa.create_engine.connect_args` dictionary::

    engine = create_engine(
        "mysql+mysqldb://scott:tiger@192.168.0.134/test",
        connect_args={
            "ssl": {
                "ca": "/home/gord/client-ssl/ca.pem",
                "cert": "/home/gord/client-ssl/client-cert.pem",
                "key": "/home/gord/client-ssl/client-key.pem",
            }
        },
    )

For convenience, the following keys may also be specified inline within the URL
where they will be interpreted into the "ssl" dictionary automatically:
"ssl_ca", "ssl_cert", "ssl_key", "ssl_capath", "ssl_cipher",
"ssl_check_hostname". An example is as follows::

    connection_uri = (
        "mysql+mysqldb://scott:tiger@192.168.0.134/test"
        "?ssl_ca=/home/gord/client-ssl/ca.pem"
        "&ssl_cert=/home/gord/client-ssl/client-cert.pem"
        "&ssl_key=/home/gord/client-ssl/client-key.pem"
    )

.. seealso::

    :ref:`pymysql_ssl` in the PyMySQL dialect


Using MySQLdb with Google Cloud SQL
-----------------------------------

Google Cloud SQL now recommends use of the MySQLdb dialect.  Connect
using a URL like the following:

.. sourcecode:: text

    mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/<projectid>:<instancename>

Server Side Cursors
-------------------

The mysqldb dialect supports server-side cursors. See :ref:`mysql_ss_cursors`.

"""

import re

from .base import MySQLCompiler
from .base import MySQLDialect
from .base import MySQLExecutionContext
from .base import MySQLIdentifierPreparer
from .base import TEXT
from ... import sql
from ... import util


class MySQLExecutionContext_mysqldb(MySQLExecutionContext):
    pass


class MySQLCompiler_mysqldb(MySQLCompiler):
    pass


class MySQLDialect_mysqldb(MySQLDialect):
    driver = "mysqldb"
    supports_statement_cache = True
    supports_unicode_statements = True
    supports_sane_rowcount = True
    supports_sane_multi_rowcount = True

    supports_native_decimal = True

    default_paramstyle = "format"
    execution_ctx_cls = MySQLExecutionContext_mysqldb
    statement_compiler = MySQLCompiler_mysqldb
    preparer = MySQLIdentifierPreparer

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._mysql_dbapi_version = (
            self._parse_dbapi_version(self.dbapi.__version__)
            if self.dbapi is not None and hasattr(self.dbapi, "__version__")
            else (0, 0, 0)
        )

    def _parse_dbapi_version(self, version):
        m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)
        if m:
            return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
        else:
            return (0, 0, 0)

    @util.langhelpers.memoized_property
    def supports_server_side_cursors(self):
        try:
            cursors = __import__("MySQLdb.cursors").cursors
            self._sscursor = cursors.SSCursor
            return True
        except (ImportError, AttributeError):
            return False

    @classmethod
    def import_dbapi(cls):
        return __import__("MySQLdb")

    def on_connect(self):
        super_ = super().on_connect()

        def on_connect(conn):
            if super_ is not None:
                super_(conn)

            charset_name = conn.character_set_name()

            if charset_name is not None:
                cursor = conn.cursor()
                cursor.execute("SET NAMES %s" % charset_name)
                cursor.close()

        return on_connect

    def do_ping(self, dbapi_connection):
        dbapi_connection.ping()
        return True

    def do_executemany(self, cursor, statement, parameters, context=None):
        rowcount = cursor.executemany(statement, parameters)
        if context is not None:
            context._rowcount = rowcount

    def _check_unicode_returns(self, connection):
        # work around issue fixed in
        # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8
        # specific issue w/ the utf8mb4_bin collation and unicode returns

        collation = connection.exec_driver_sql(
            "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"
            % (
                self.identifier_preparer.quote("Charset"),
                self.identifier_preparer.quote("Collation"),
            )
        ).scalar()
        has_utf8mb4_bin = self.server_version_info > (5,) and collation
        if has_utf8mb4_bin:
            additional_tests = [
                sql.collate(
                    sql.cast(
                        sql.literal_column("'test collated returns'"),
                        TEXT(charset="utf8mb4"),
                    ),
                    "utf8mb4_bin",
                )
            ]
        else:
            additional_tests = []
        return super()._check_unicode_returns(connection, additional_tests)

    def create_connect_args(self, url, _translate_args=None):
        if _translate_args is None:
            _translate_args = dict(
                database="db", username="user", password="passwd"
            )

        opts = url.translate_connect_args(**_translate_args)
        opts.update(url.query)

        util.coerce_kw_type(opts, "compress", bool)
        util.coerce_kw_type(opts, "connect_timeout", int)
        util.coerce_kw_type(opts, "read_timeout", int)
        util.coerce_kw_type(opts, "write_timeout", int)
        util.coerce_kw_type(opts, "client_flag", int)
        util.coerce_kw_type(opts, "local_infile", bool)
        # Note: using either of the below will cause all strings to be
        # returned as Unicode, both in raw SQL operations and with column
        # types like String and MSString.
        util.coerce_kw_type(opts, "use_unicode", bool)
        util.coerce_kw_type(opts, "charset", str)

        # Rich values 'cursorclass' and 'conv' are not supported via
        # query string.

        ssl = {}
        keys = [
            ("ssl_ca", str),
            ("ssl_key", str),
            ("ssl_cert", str),
            ("ssl_capath", str),
            ("ssl_cipher", str),
            ("ssl_check_hostname", bool),
        ]
        for key, kw_type in keys:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], kw_type)
                del opts[key]
        if ssl:
            opts["ssl"] = ssl

        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
        # supports_sane_rowcount.
        client_flag = opts.get("client_flag", 0)

        client_flag_found_rows = self._found_rows_client_flag()
        if client_flag_found_rows is not None:
            client_flag |= client_flag_found_rows
            opts["client_flag"] = client_flag
        return [[], opts]

    def _found_rows_client_flag(self):
        if self.dbapi is not None:
            try:
                CLIENT_FLAGS = __import__(
                    self.dbapi.__name__ + ".constants.CLIENT"
                ).constants.CLIENT
            except (AttributeError, ImportError):
                return None
            else:
                return CLIENT_FLAGS.FOUND_ROWS
        else:
            return None

    def _extract_error_code(self, exception):
        return exception.args[0]

    def _detect_charset(self, connection):
        """Sniff out the character set in use for connection results."""

        try:
            # note: the SQL here would be
            # "SHOW VARIABLES LIKE 'character_set%%'"
            cset_name = connection.connection.character_set_name
        except AttributeError:
            util.warn(
                "No 'character_set_name' can be detected with "
                "this MySQL-Python version; "
                "please upgrade to a recent version of MySQL-Python.  "
                "Assuming latin1."
            )
            return "latin1"
        else:
            return cset_name()

    def get_isolation_level_values(self, dbapi_connection):
        return (
            "SERIALIZABLE",
            "READ UNCOMMITTED",
            "READ COMMITTED",
            "REPEATABLE READ",
            "AUTOCOMMIT",
        )

    def set_isolation_level(self, dbapi_connection, level):
        if level == "AUTOCOMMIT":
            dbapi_connection.autocommit(True)
        else:
            dbapi_connection.autocommit(False)
            super().set_isolation_level(dbapi_connection, level)


dialect = MySQLDialect_mysqldb