Zy Filemanager

📁 File Browser

Name	Size	Permissions	Actions
📁 __pycache__	DIR	0755	🗑️
📄 __init__.py (2,056 bytes)	2.01 KB	0644	✍️ ⬇️ 🗑️
📄 base.py (107,282 bytes)	104.77 KB	0644	✍️ ⬇️ 🗑️
📄 cymysql.py (2,244 bytes)	2.19 KB	0644	✍️ ⬇️ 🗑️
📄 dml.py (4,764 bytes)	4.65 KB	0644	✍️ ⬇️ 🗑️
📄 enumerated.py (11,307 bytes)	11.04 KB	0644	✍️ ⬇️ 🗑️
📄 gaerdbms.py (3,375 bytes)	3.3 KB	0644	✍️ ⬇️ 🗑️
📄 json.py (2,045 bytes)	2 KB	0644	✍️ ⬇️ 🗑️
📄 mysqlconnector.py (7,889 bytes)	7.7 KB	0644	✍️ ⬇️ 🗑️
📄 mysqldb.py (8,504 bytes)	8.3 KB	0644	✍️ ⬇️ 🗑️
📄 oursql.py (8,086 bytes)	7.9 KB	0644	✍️ ⬇️ 🗑️
📄 provision.py (1,269 bytes)	1.24 KB	0644	✍️ ⬇️ 🗑️
📄 pymysql.py (2,720 bytes)	2.66 KB	0644	✍️ ⬇️ 🗑️
📄 pyodbc.py (3,470 bytes)	3.39 KB	0644	✍️ ⬇️ 🗑️
📄 reflection.py (18,275 bytes)	17.85 KB	0644	✍️ ⬇️ 🗑️
📄 types.py (24,589 bytes)	24.01 KB	0644	✍️ ⬇️ 🗑️
📄 zxjdbc.py (3,970 bytes)	3.88 KB	0644	✍️ ⬇️ 🗑️

✍️ Editing: zxjdbc.py

            File: /opt/cloudlinux/venv/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/zxjdbc.py

            Size: 3.88 KB | 
            Last Modified: 2026-01-20 20:01:47        

# 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

↩️ Back

Zy Filemanager

📁 File Browser

📤 Upload Files

Single File

Multiple Files

ZIP Extract

➕ Create New

New Folder

New File

✍️ Editing: zxjdbc.py

Backup Management