access.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

# access.py
# Copyright (C) 2007 Paul Johnston, [e-mail]
# Portions derived from jet2sql.py by Matt Keranen, [e-mail]
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php

from sqlalchemy import sql, schema, types, exceptions, pool
from sqlalchemy.sql import compiler, expression
from sqlalchemy.engine import default, base


class AcNumeric(types.Numeric):
    def result_processor(self, dialect):
        return None

    def bind_processor(self, dialect):
        def process(value):
            if value is None:
                # Not sure that this exception is needed
                return value
            else:
                return str(value)
        return process
        
    def get_col_spec(self):
        return "NUMERIC"

class AcFloat(types.Float):
    def get_col_spec(self):
        return "FLOAT"

    def bind_processor(self, dialect):
        """By converting to string, we can use Decimal types round-trip."""
        def process(value):
            if not value is None:
                return str(value)
            return None
        return process
        
class AcInteger(types.Integer):
    def get_col_spec(self):
        return "INTEGER"

class AcTinyInteger(types.Integer):
    def get_col_spec(self):
        return "TINYINT"

class AcSmallInteger(types.Smallinteger):
    def get_col_spec(self):
        return "SMALLINT"

class AcDateTime(types.DateTime):
    def __init__(self, *a, **kw):
        super(AcDateTime, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

class AcDate(types.Date):
    def __init__(self, *a, **kw):
        super(AcDate, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

class AcText(types.Text):
    def get_col_spec(self):
        return "MEMO"

class AcString(types.String):
    def get_col_spec(self):
        return "TEXT" + (self.length and ("(%d)" % self.length) or "")

class AcUnicode(types.Unicode):
    def get_col_spec(self):
        return "TEXT" + (self.length and ("(%d)" % self.length) or "")

    def bind_processor(self, dialect):
        return None

    def result_processor(self, dialect):
        return None

class AcChar(types.CHAR):
    def get_col_spec(self):        
        return "TEXT" + (self.length and ("(%d)" % self.length) or "")

class AcBinary(types.Binary):
    def get_col_spec(self):
        return "BINARY"

class AcBoolean(types.Boolean):
    def get_col_spec(self):
        return "YESNO"

    def result_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            return value and True or False
        return process
        
    def bind_processor(self, dialect):
        def process(value):
            if value is True:
                return 1
            elif value is False:
                return 0
            elif value is None:
                return None
            else:
                return value and True or False
        return process
        
class AcTimeStamp(types.TIMESTAMP):
    def get_col_spec(self):
        return "TIMESTAMP"

def descriptor():
    return {'name':'access',
    'description':'Microsoft Access',
    'arguments':[
        ('user',"Database user name",None),
        ('password',"Database password",None),
        ('db',"Path to database file",None),
    ]}

class AccessExecutionContext(default.DefaultExecutionContext):
    def _has_implicit_sequence(self, column):
        if column.primary_key and column.autoincrement:
            if isinstance(column.type, types.Integer) and not column.foreign_key:
                if column.default is None or (isinstance(column.default, schema.Sequence) and \
                                              column.default.optional):
                    return True
        return False

    def post_exec(self):
        """If we inserted into a row with a COUNTER column, fetch the ID"""

        if self.compiled.isinsert:
            tbl = self.compiled.statement.table
            if not hasattr(tbl, 'has_sequence'):
                tbl.has_sequence = None
                for column in tbl.c:
                    if getattr(column, 'sequence', False) or self._has_implicit_sequence(column):
                        tbl.has_sequence = column
                        break

            if bool(tbl.has_sequence):
                # TBD: for some reason _last_inserted_ids doesn't exist here
                # (but it does at corresponding point in mssql???)
                #if not len(self._last_inserted_ids) or self._last_inserted_ids[0] is None:
                self.cursor.execute("SELECT @@identity AS lastrowid")
                row = self.cursor.fetchone()
                self._last_inserted_ids = [int(row[0])] #+ self._last_inserted_ids[1:]
                # print "LAST ROW ID", self._last_inserted_ids

        super(AccessExecutionContext, self).post_exec()


const, daoEngine = None, None
class AccessDialect(default.DefaultDialect):
    colspecs = {
        types.Unicode : AcUnicode,
        types.Integer : AcInteger,
        types.Smallinteger: AcSmallInteger,
        types.Numeric : AcNumeric,
        types.Float : AcFloat,
        types.DateTime : AcDateTime,
        types.Date : AcDate,
        types.String : AcString,
        types.Binary : AcBinary,
        types.Boolean : AcBoolean,
        types.Text : AcText,
        types.CHAR: AcChar,
        types.TIMESTAMP: AcTimeStamp,
    }

    supports_sane_rowcount = False
    supports_sane_multi_rowcount = False

    def type_descriptor(self, typeobj):
        newobj = types.adapt_type(typeobj, self.colspecs)
        return newobj

    def __init__(self, **params):
        super(AccessDialect, self).__init__(**params)
        self.text_as_varchar = False
        self._dtbs = None

    def dbapi(cls):
        import win32com.client, pythoncom

        global const, daoEngine
        if const is None:
            const = win32com.client.constants
            for suffix in (".36", ".35", ".30"):
                try:
                    daoEngine = win32com.client.gencache.EnsureDispatch("DAO.DBEngine" + suffix)
                    break
                except pythoncom.com_error:
                    pass
            else:
                raise exceptions.InvalidRequestError("Can't find a DB engine. Check http://support.microsoft.com/kb/239114 for details.")

        import pyodbc as module
        return module
    dbapi = classmethod(dbapi)

    def create_connect_args(self, url):
        opts = url.translate_connect_args()
        connectors = ["Driver={Microsoft Access Driver (*.mdb)}"]
        connectors.append("Dbq=%s" % opts["database"])
        user = opts.get("username", None)
        if user: