# 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

"""

.. 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`.

"""
from __future__ import annotations

import re
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Optional
from typing import Tuple
from typing import TYPE_CHECKING

from .base import MySQLCompiler
from .base import MySQLDialect
from .base import MySQLExecutionContext
from .base import MySQLIdentifierPreparer
from ... import util
from ...util.typing import Literal

if TYPE_CHECKING:

    from ...engine.base import Connection
    from ...engine.interfaces import _DBAPIMultiExecuteParams
    from ...engine.interfaces import ConnectArgsType
    from ...engine.interfaces import DBAPIConnection
    from ...engine.interfaces import DBAPICursor
    from ...engine.interfaces import DBAPIModule
    from ...engine.interfaces import ExecutionContext
    from ...engine.interfaces import IsolationLevel
    from ...engine.url import URL


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
    server_version_info: Tuple[int, ...]

    def __init__(self, **kwargs: Any):
        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: str) -> Tuple[int, ...]:
        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) -> bool:
        try:
            cursors = __import__("MySQLdb.cursors").cursors
            self._sscursor = cursors.SSCursor
            return True
        except (ImportError, AttributeError):
            return False

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

    def on_connect(self) -> Callable[[DBAPIConnection], None]:
        super_ = super().on_connect()

        def on_connect(conn: DBAPIConnection) -> None:
            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: DBAPIConnection) -> Literal[True]:
        dbapi_connection.ping()
        return True

    def do_executemany(
        self,
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPIMultiExecuteParams,
        context: Optional[ExecutionContext] = None,
    ) -> None:
        rowcount = cursor.executemany(statement, parameters)
        if context is not None:
            cast(MySQLExecutionContext, context)._rowcount = rowcount

    def create_connect_args(
        self, url: URL, _translate_args: Optional[Dict[str, Any]] = None
    ) -> ConnectArgsType:
        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) -> Optional[int]:
        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  # type: ignore
        else:
            return None

    def _extract_error_code(self, exception: DBAPIModule.Error) -> int:
        return exception.args[0]  # type: ignore[no-any-return]

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

        try:
            # note: the SQL here would be
            # "SHOW VARIABLES LIKE 'character_set%%'"

            cset_name: Callable[[], str] = (
                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_conn: DBAPIConnection
    ) -> Tuple[IsolationLevel, ...]:
        return (
            "SERIALIZABLE",
            "READ UNCOMMITTED",
            "READ COMMITTED",
            "REPEATABLE READ",
            "AUTOCOMMIT",
        )

    def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
        return dbapi_conn.get_autocommit()  # type: ignore[no-any-return]

    def set_isolation_level(
        self, dbapi_connection: DBAPIConnection, level: IsolationLevel
    ) -> None:
        if level == "AUTOCOMMIT":
            dbapi_connection.autocommit(True)
        else:
            dbapi_connection.autocommit(False)
            super().set_isolation_level(dbapi_connection, level)


dialect = MySQLDialect_mysqldb