✏️ Editing: mysqldb.py Read Only

# mysql/mysqldb.py
# Copyright (C) 2005-2021 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`.

"""

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

def on_connect(self):
        super_ = super(MySQLDialect_mysqldb, self).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):
        try:
            dbapi_connection.ping(False)
        except self.dbapi.Error as err:
            if self.is_disconnect(err, dbapi_connection, None):
                return False
            else:
                raise
        else:
            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

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, _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", 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

🔐 Sid Gifari File Manager Pro