zxjdbc.py :  » Database » SQLAlchemy » SQLAlchemy-0.6.0 » lib » sqlalchemy » dialects » mysql » Python Open Source

Home
Python Open Source
1.3.1.2 Python
2.Ajax
3.Aspect Oriented
4.Blog
5.Build
6.Business Application
7.Chart Report
8.Content Management Systems
9.Cryptographic
10.Database
11.Development
12.Editor
13.Email
14.ERP
15.Game 2D 3D
16.GIS
17.GUI
18.IDE
19.Installer
20.IRC
21.Issue Tracker
22.Language Interface
23.Log
24.Math
25.Media Sound Audio
26.Mobile
27.Network
28.Parser
29.PDF
30.Project Management
31.RSS
32.Search
33.Security
34.Template Engines
35.Test
36.UML
37.USB Serial
38.Web Frameworks
39.Web Server
40.Web Services
41.Web Unit
42.Wiki
43.Windows
44.XML
Python Open Source » Database » SQLAlchemy 
SQLAlchemy » SQLAlchemy 0.6.0 » lib » sqlalchemy » dialects » mysql » zxjdbc.py
"""Support for the MySQL database via Jython's zxjdbc JDBC connector.

JDBC Driver
-----------

The official MySQL JDBC driver is at
http://dev.mysql.com/downloads/connector/j/.

Connecting
----------

Connect string format:

    mysql+zxjdbc://<user>:<password>@<hostname>[:<port>]/<database>

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
overriden via a ``create_engine`` URL parameter.

"""
import re

from sqlalchemy import types
from sqlalchemy.connectors.zxJDBC import ZxJDBCConnector
from sqlalchemy.dialects.mysql.base import BIT,MySQLDialect,MySQLExecutionContext

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 = 0L
            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 = dict((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.orig.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('[.\-]')
        for n in r.split(dbapi_con.dbversion):
            try:
                version.append(int(n))
            except ValueError:
                version.append(n)
        return tuple(version)

dialect = MySQLDialect_zxjdbc
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.