# mysql/mysqldb.py
# Copyright (C) 2005-2019 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://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 <http://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.


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

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

    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 .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):
    @property
    def rowcount(self):
        if hasattr(self, "_rowcount"):
            return self._rowcount
        else:
            return self.cursor.rowcount


class MySQLCompiler_mysqldb(MySQLCompiler):
    pass


class MySQLIdentifierPreparer_mysqldb(MySQLIdentifierPreparer):
    pass


class MySQLDialect_mysqldb(MySQLDialect):
    driver = "mysqldb"
    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_mysqldb

    def __init__(self, server_side_cursors=False, **kwargs):
        super(MySQLDialect_mysqldb, self).__init__(**kwargs)
        self.server_side_cursors = server_side_cursors

    @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 dbapi(cls):
        return __import__("MySQLdb")

    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

        has_utf8mb4_bin = self.server_version_info > (
            5,
        ) and connection.scalar(
            "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"
            % (
                self.identifier_preparer.quote("Charset"),
                self.identifier_preparer.quote("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(MySQLDialect_mysqldb, self)._check_unicode_returns(
            connection, additional_tests
        )

    def create_connect_args(self, url):
        opts = url.translate_connect_args(
            database="db", username="user", password="passwd"
        )
        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", int)
        # 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", "ssl_key", "ssl_cert", "ssl_capath", "ssl_cipher"]
        for key in keys:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], str)
                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)
        if self.dbapi is not None:
            try:
                CLIENT_FLAGS = __import__(
                    self.dbapi.__name__ + ".constants.CLIENT"
                ).constants.CLIENT
                client_flag |= CLIENT_FLAGS.FOUND_ROWS
            except (AttributeError, ImportError):
                self.supports_sane_rowcount = False
            opts["client_flag"] = client_flag
        return [[], opts]

    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()

    _isolation_lookup = set(
        [
            "SERIALIZABLE",
            "READ UNCOMMITTED",
            "READ COMMITTED",
            "REPEATABLE READ",
            "AUTOCOMMIT",
        ]
    )

    def _set_isolation_level(self, connection, level):
        if level == "AUTOCOMMIT":
            connection.autocommit(True)
        else:
            connection.autocommit(False)
            super(MySQLDialect_mysqldb, self)._set_isolation_level(
                connection, level
            )


dialect = MySQLDialect_mysqldb