Viewing file: mysqldb.py (8.3 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
# 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
|