diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py | 303 |
1 files changed, 0 insertions, 303 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py b/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py deleted file mode 100644 index 0c632b6..0000000 --- a/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py +++ /dev/null @@ -1,303 +0,0 @@ -# dialects/mysql/mysqldb.py -# Copyright (C) 2005-2024 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:: - - 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", 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", 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 |