venv

1 files changed, 303 insertions, 0 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
new file mode 100644
index 0000000..0c632b6
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/mysqldb.py
@@ -0,0 +1,303 @@
+# 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