mxodbc.py :  » Database » SQLAlchemy » SQLAlchemy-0.6.0 » lib » sqlalchemy » connectors » 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 » connectors » mxodbc.py
"""
Provide an SQLALchemy connector for the eGenix mxODBC commercial
Python adapter for ODBC. This is not a free product, but eGenix
provides SQLAlchemy with a license for use in continuous integration
testing.

This has been tested for use with mxODBC 3.1.2 on SQL Server 2005
and 2008, using the SQL Server Native driver. However, it is
possible for this to be used on other database platforms.

For more info on mxODBC, see http://www.egenix.com/

"""

import sys
import re
import warnings
from decimal import Decimal

from sqlalchemy.connectors import Connector
from sqlalchemy import types
import sqlalchemy.processors as processors

class MxODBCConnector(Connector):
    driver='mxodbc'
    
    supports_sane_multi_rowcount = False
    supports_unicode_statements = False
    supports_unicode_binds = False
    
    supports_native_decimal = True
    
    @classmethod
    def dbapi(cls):
        # this classmethod will normally be replaced by an instance
        # attribute of the same name, so this is normally only called once.
        cls._load_mx_exceptions()
        platform = sys.platform
        if platform == 'win32':
            from mx.ODBC import Windows
        # this can be the string "linux2", and possibly others
        elif 'linux' in platform:
            from mx.ODBC import unixODBC
        elif platform == 'darwin':
            from mx.ODBC import iODBC
        else:
            raise ImportError, "Unrecognized platform for mxODBC import"
        return module

    @classmethod
    def _load_mx_exceptions(cls):
        """ Import mxODBC exception classes into the module namespace,
        as if they had been imported normally. This is done here
        to avoid requiring all SQLAlchemy users to install mxODBC.
        """
        global InterfaceError, ProgrammingError
        from mx.ODBC import InterfaceError
        from mx.ODBC import ProgrammingError

    def on_connect(self):
        def connect(conn):
            conn.stringformat = self.dbapi.MIXED_STRINGFORMAT
            conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT
            conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT
            conn.errorhandler = self._error_handler()
        return connect
    
    def _error_handler(self):
        """ Return a handler that adjusts mxODBC's raised Warnings to
        emit Python standard warnings.
        """
        from mx.ODBC.Error import Warning
        def error_handler(connection, cursor, errorclass, errorvalue):

            if issubclass(errorclass, MxOdbcWarning):
                errorclass.__bases__ = (Warning,)
                warnings.warn(message=str(errorvalue),
                          category=errorclass,
                          stacklevel=2)
            else:
                raise errorclass, errorvalue
        return error_handler

    def create_connect_args(self, url):
        """ Return a tuple of *args,**kwargs for creating a connection.

        The mxODBC 3.x connection constructor looks like this:

            connect(dsn, user='', password='',
                    clear_auto_commit=1, errorhandler=None)

        This method translates the values in the provided uri
        into args and kwargs needed to instantiate an mxODBC Connection.

        The arg 'errorhandler' is not used by SQLAlchemy and will
        not be populated.
        
        """
        opts = url.translate_connect_args(username='user')
        opts.update(url.query)
        args = opts.pop('host')
        opts.pop('port', None)
        opts.pop('database', None)
        return (args,), opts

    def is_disconnect(self, e):
        # eGenix recommends checking connection.closed here,
        # but how can we get a handle on the current connection?
        if isinstance(e, self.dbapi.ProgrammingError):
            return "connection already closed" in str(e)
        elif isinstance(e, self.dbapi.Error):
            return '[08S01]' in str(e)
        else:
            return False

    def _get_server_version_info(self, connection):
        # eGenix suggests using conn.dbms_version instead of what we're doing here
        dbapi_con = connection.connection
        version = []
        r = re.compile('[.\-]')
        # 18 == pyodbc.SQL_DBMS_VER
        for n in r.split(dbapi_con.getinfo(18)[1]):
            try:
                version.append(int(n))
            except ValueError:
                version.append(n)
        return tuple(version)

    def do_execute(self, cursor, statement, parameters, context=None):
        if context:
            native_odbc_execute = context.execution_options.\
                                        get('native_odbc_execute', 'auto')
            if native_odbc_execute is True:
                # user specified native_odbc_execute=True
                cursor.execute(statement, parameters)
            elif native_odbc_execute is False:
                # user specified native_odbc_execute=False
                cursor.executedirect(statement, parameters)
            elif context.is_crud:
                # statement is UPDATE, DELETE, INSERT
                cursor.execute(statement, parameters)
            else:
                # all other statements
                cursor.executedirect(statement, parameters)
        else:
            cursor.executedirect(statement, parameters)
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.