File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql

opt / cloudlinux / venv / lib / python3.11 / site-packages / sqlalchemy / dialects / mysql /

?? ..
?? __init__.py(2.01 KB)
?? __pycache__
?? base.py(104.77 KB)
?? cymysql.py(2.19 KB)
?? dml.py(4.65 KB)
?? enumerated.py(11.04 KB)
?? gaerdbms.py(3.3 KB)
?? json.py(2 KB)
?? mysqlconnector.py(7.7 KB)
?? mysqldb.py(8.3 KB)
?? oursql.py(7.9 KB)
?? provision.py(1.24 KB)
?? pymysql.py(2.66 KB)
?? pyodbc.py(3.39 KB)
?? reflection.py(17.85 KB)
?? types.py(24.01 KB)
?? zxjdbc.py(3.88 KB)

Editing: zxjdbc.py

# mysql/zxjdbc.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

r"""

.. dialect:: mysql+zxjdbc
    :name: zxjdbc for Jython
    :dbapi: zxjdbc
    :connectstring: mysql+zxjdbc://<user>:<password>@<hostname>[:<port>]/<database>
    :driverurl: http://dev.mysql.com/downloads/connector/j/

.. note:: Jython is not supported by current versions of SQLAlchemy.  The
       zxjdbc dialect should be considered as experimental.

Character Sets
--------------

SQLAlchemy zxjdbc dialects pass unicode straight through to the
zxjdbc/JDBC layer. To allow multiple character sets to be sent from the
MySQL Connector/J JDBC driver, by default SQLAlchemy sets its
``characterEncoding`` connection property to ``UTF-8``. It may be
overridden via a ``create_engine`` URL parameter.

"""  # noqa
import re

from .base import BIT
from .base import MySQLDialect
from .base import MySQLExecutionContext
from ... import types as sqltypes
from ... import util
from ...connectors.zxJDBC import ZxJDBCConnector

class _ZxJDBCBit(BIT):
    def result_processor(self, dialect, coltype):
        """Converts boolean or byte arrays from MySQL Connector/J to longs."""

def process(value):
            if value is None:
                return value
            if isinstance(value, bool):
                return int(value)
            v = 0
            for i in value:
                v = v << 8 | (i & 0xFF)
            value = v
            return value

return process

class MySQLExecutionContext_zxjdbc(MySQLExecutionContext):
    def get_lastrowid(self):
        cursor = self.create_cursor()
        cursor.execute("SELECT LAST_INSERT_ID()")
        lastrowid = cursor.fetchone()[0]
        cursor.close()
        return lastrowid

class MySQLDialect_zxjdbc(ZxJDBCConnector, MySQLDialect):
    jdbc_db_name = "mysql"
    jdbc_driver_name = "com.mysql.jdbc.Driver"

execution_ctx_cls = MySQLExecutionContext_zxjdbc

colspecs = util.update_copy(
        MySQLDialect.colspecs, {sqltypes.Time: sqltypes.Time, BIT: _ZxJDBCBit}
    )

def _detect_charset(self, connection):
        """Sniff out the character set in use for connection results."""
        # Prefer 'character_set_results' for the current connection over the
        # value in the driver.  SET NAMES or individual variable SETs will
        # change the charset without updating the driver's view of the world.
        #
        # If it's decided that issuing that sort of SQL leaves you SOL, then
        # this can prefer the driver value.
        rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
        opts = {row[0]: row[1] for row in self._compat_fetchall(rs)}
        for key in ("character_set_connection", "character_set"):
            if opts.get(key, None):
                return opts[key]

util.warn(
            "Could not detect the connection character set.  "
            "Assuming latin1."
        )
        return "latin1"

def _driver_kwargs(self):
        """return kw arg dict to be sent to connect()."""
        return dict(characterEncoding="UTF-8", yearIsDateType="false")

def _extract_error_code(self, exception):
        # e.g.: DBAPIError: (Error) Table 'test.u2' doesn't exist
        # [SQLCode: 1146], [SQLState: 42S02] 'DESCRIBE `u2`' ()
        m = re.compile(r"\[SQLCode\: (\d+)\]").search(str(exception.args))
        c = m.group(1)
        if c:
            return int(c)

def _get_server_version_info(self, connection):
        dbapi_con = connection.connection
        version = []
        r = re.compile(r"[.\-]")
        for n in r.split(dbapi_con.dbversion):
            try:
                version.append(int(n))
            except ValueError:
                version.append(n)
        return tuple(version)

dialect = MySQLDialect_zxjdbc

X7ROOT File Manager

Editing: zxjdbc.py

Upload File

Create Folder