1# mysql/zxjdbc.py 2# Copyright (C) 2005-2016 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 8""" 9 10.. dialect:: mysql+zxjdbc 11 :name: zxjdbc for Jython 12 :dbapi: zxjdbc 13 :connectstring: mysql+zxjdbc://<user>:<password>@<hostname>[:<port>]/\ 14<database> 15 :driverurl: http://dev.mysql.com/downloads/connector/j/ 16 17 .. note:: Jython is not supported by current versions of SQLAlchemy. The 18 zxjdbc dialect should be considered as experimental. 19 20Character Sets 21-------------- 22 23SQLAlchemy zxjdbc dialects pass unicode straight through to the 24zxjdbc/JDBC layer. To allow multiple character sets to be sent from the 25MySQL Connector/J JDBC driver, by default SQLAlchemy sets its 26``characterEncoding`` connection property to ``UTF-8``. It may be 27overridden via a ``create_engine`` URL parameter. 28 29""" 30import re 31 32from ... import types as sqltypes, util 33from ...connectors.zxJDBC import ZxJDBCConnector 34from .base import BIT, MySQLDialect, MySQLExecutionContext 35 36 37class _ZxJDBCBit(BIT): 38 def result_processor(self, dialect, coltype): 39 """Converts boolean or byte arrays from MySQL Connector/J to longs.""" 40 def process(value): 41 if value is None: 42 return value 43 if isinstance(value, bool): 44 return int(value) 45 v = 0 46 for i in value: 47 v = v << 8 | (i & 0xff) 48 value = v 49 return value 50 return process 51 52 53class MySQLExecutionContext_zxjdbc(MySQLExecutionContext): 54 def get_lastrowid(self): 55 cursor = self.create_cursor() 56 cursor.execute("SELECT LAST_INSERT_ID()") 57 lastrowid = cursor.fetchone()[0] 58 cursor.close() 59 return lastrowid 60 61 62class MySQLDialect_zxjdbc(ZxJDBCConnector, MySQLDialect): 63 jdbc_db_name = 'mysql' 64 jdbc_driver_name = 'com.mysql.jdbc.Driver' 65 66 execution_ctx_cls = MySQLExecutionContext_zxjdbc 67 68 colspecs = util.update_copy( 69 MySQLDialect.colspecs, 70 { 71 sqltypes.Time: sqltypes.Time, 72 BIT: _ZxJDBCBit 73 } 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 = dict((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("Could not detect the connection character set. " 91 "Assuming latin1.") 92 return 'latin1' 93 94 def _driver_kwargs(self): 95 """return kw arg dict to be sent to connect().""" 96 return dict(characterEncoding='UTF-8', yearIsDateType='false') 97 98 def _extract_error_code(self, exception): 99 # e.g.: DBAPIError: (Error) Table 'test.u2' doesn't exist 100 # [SQLCode: 1146], [SQLState: 42S02] 'DESCRIBE `u2`' () 101 m = re.compile(r"\[SQLCode\: (\d+)\]").search(str(exception.args)) 102 c = m.group(1) 103 if c: 104 return int(c) 105 106 def _get_server_version_info(self, connection): 107 dbapi_con = connection.connection 108 version = [] 109 r = re.compile('[.\-]') 110 for n in r.split(dbapi_con.dbversion): 111 try: 112 version.append(int(n)) 113 except ValueError: 114 version.append(n) 115 return tuple(version) 116 117dialect = MySQLDialect_zxjdbc 118