1# mysql/zxjdbc.py 2# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors 3# <see AUTHORS file> 4# 5# This module is part of SQLAlchemy and is released under 6# the MIT License: http://www.opensource.org/licenses/mit-license.php 7 8r""" 9 10.. dialect:: mysql+zxjdbc 11 :name: zxjdbc for Jython 12 :dbapi: zxjdbc 13 :connectstring: mysql+zxjdbc://<user>:<password>@<hostname>[:<port>]/<database> 14 :driverurl: http://dev.mysql.com/downloads/connector/j/ 15 16 .. note:: Jython is not supported by current versions of SQLAlchemy. The 17 zxjdbc dialect should be considered as experimental. 18 19Character Sets 20-------------- 21 22SQLAlchemy zxjdbc dialects pass unicode straight through to the 23zxjdbc/JDBC layer. To allow multiple character sets to be sent from the 24MySQL Connector/J JDBC driver, by default SQLAlchemy sets its 25``characterEncoding`` connection property to ``UTF-8``. It may be 26overridden via a ``create_engine`` URL parameter. 27 28""" # noqa 29import re 30 31from .base import BIT 32from .base import MySQLDialect 33from .base import MySQLExecutionContext 34from ... import types as sqltypes 35from ... import util 36from ...connectors.zxJDBC import ZxJDBCConnector 37 38 39class _ZxJDBCBit(BIT): 40 def result_processor(self, dialect, coltype): 41 """Converts boolean or byte arrays from MySQL Connector/J to longs.""" 42 43 def process(value): 44 if value is None: 45 return value 46 if isinstance(value, bool): 47 return int(value) 48 v = 0 49 for i in value: 50 v = v << 8 | (i & 0xFF) 51 value = v 52 return value 53 54 return process 55 56 57class MySQLExecutionContext_zxjdbc(MySQLExecutionContext): 58 def get_lastrowid(self): 59 cursor = self.create_cursor() 60 cursor.execute("SELECT LAST_INSERT_ID()") 61 lastrowid = cursor.fetchone()[0] 62 cursor.close() 63 return lastrowid 64 65 66class MySQLDialect_zxjdbc(ZxJDBCConnector, MySQLDialect): 67 jdbc_db_name = "mysql" 68 jdbc_driver_name = "com.mysql.jdbc.Driver" 69 70 execution_ctx_cls = MySQLExecutionContext_zxjdbc 71 72 colspecs = util.update_copy( 73 MySQLDialect.colspecs, {sqltypes.Time: sqltypes.Time, BIT: _ZxJDBCBit} 74 ) 75 76 def _detect_charset(self, connection): 77 """Sniff out the character set in use for connection results.""" 78 # Prefer 'character_set_results' for the current connection over the 79 # value in the driver. SET NAMES or individual variable SETs will 80 # change the charset without updating the driver's view of the world. 81 # 82 # If it's decided that issuing that sort of SQL leaves you SOL, then 83 # this can prefer the driver value. 84 rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'") 85 opts = {row[0]: row[1] for row in self._compat_fetchall(rs)} 86 for key in ("character_set_connection", "character_set"): 87 if opts.get(key, None): 88 return opts[key] 89 90 util.warn( 91 "Could not detect the connection character set. " 92 "Assuming latin1." 93 ) 94 return "latin1" 95 96 def _driver_kwargs(self): 97 """return kw arg dict to be sent to connect().""" 98 return dict(characterEncoding="UTF-8", yearIsDateType="false") 99 100 def _extract_error_code(self, exception): 101 # e.g.: DBAPIError: (Error) Table 'test.u2' doesn't exist 102 # [SQLCode: 1146], [SQLState: 42S02] 'DESCRIBE `u2`' () 103 m = re.compile(r"\[SQLCode\: (\d+)\]").search(str(exception.args)) 104 c = m.group(1) 105 if c: 106 return int(c) 107 108 def _get_server_version_info(self, connection): 109 dbapi_con = connection.connection 110 version = [] 111 r = re.compile(r"[.\-]") 112 for n in r.split(dbapi_con.dbversion): 113 try: 114 version.append(int(n)) 115 except ValueError: 116 version.append(n) 117 return tuple(version) 118 119 120dialect = MySQLDialect_zxjdbc 121