Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/mysql/zxjdbc.py : 37%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# mysql/zxjdbc.py
2# Copyright (C) 2005-2020 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
8r"""
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/
16 .. note:: Jython is not supported by current versions of SQLAlchemy. The
17 zxjdbc dialect should be considered as experimental.
19Character Sets
20--------------
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.
28""" # noqa
29import re
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
39class _ZxJDBCBit(BIT):
40 def result_processor(self, dialect, coltype):
41 """Converts boolean or byte arrays from MySQL Connector/J to longs."""
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
54 return process
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
66class MySQLDialect_zxjdbc(ZxJDBCConnector, MySQLDialect):
67 jdbc_db_name = "mysql"
68 jdbc_driver_name = "com.mysql.jdbc.Driver"
70 execution_ctx_cls = MySQLExecutionContext_zxjdbc
72 colspecs = util.update_copy(
73 MySQLDialect.colspecs, {sqltypes.Time: sqltypes.Time, BIT: _ZxJDBCBit}
74 )
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]
90 util.warn(
91 "Could not detect the connection character set. "
92 "Assuming latin1."
93 )
94 return "latin1"
96 def _driver_kwargs(self):
97 """return kw arg dict to be sent to connect()."""
98 return dict(characterEncoding="UTF-8", yearIsDateType="false")
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)
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)
120dialect = MySQLDialect_zxjdbc