sybase.py

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

# sybase.py
# Copyright (C) 2007 Fisch Asset Management AG http://www.fam.ch
# Coding: Alexander Houben [e-mail]
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php

"""
Sybase database backend.

Known issues / TODO:

 * Uses the mx.ODBC driver from egenix (version 2.1.0)
 * The current version of sqlalchemy.databases.sybase only supports
   mx.ODBC.Windows (other platforms such as mx.ODBC.unixODBC still need
   some development)
 * Support for pyodbc has been built in but is not yet complete (needs
   further development)
 * Results of running tests/alltests.py:
     Ran 934 tests in 287.032s
     FAILED (failures=3, errors=1)
 * Tested on 'Adaptive Server Anywhere 9' (version 9.0.1.1751)
"""

import datetime, operator

from sqlalchemy import util, sql, schema, exceptions
from sqlalchemy.sql import compiler, expression
from sqlalchemy.engine import default, base
from sqlalchemy import types as sqltypes
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy import MetaData, Table, Column
from sqlalchemy import String, Integer, SMALLINT, CHAR, ForeignKey


__all__ = [
    'SybaseTypeError'
    'SybaseNumeric', 'SybaseFloat', 'SybaseInteger', 'SybaseBigInteger',
    'SybaseTinyInteger', 'SybaseSmallInteger',
    'SybaseDateTime_mxodbc', 'SybaseDateTime_pyodbc',
    'SybaseDate_mxodbc', 'SybaseDate_pyodbc',
    'SybaseTime_mxodbc', 'SybaseTime_pyodbc',
    'SybaseText', 'SybaseString', 'SybaseChar', 'SybaseBinary',
    'SybaseBoolean', 'SybaseTimeStamp', 'SybaseMoney', 'SybaseSmallMoney',
    'SybaseUniqueIdentifier',
    ]


RESERVED_WORDS = util.Set([
    "add", "all", "alter", "and",
    "any", "as", "asc", "backup",
    "begin", "between", "bigint", "binary",
    "bit", "bottom", "break", "by",
    "call", "capability", "cascade", "case",
    "cast", "char", "char_convert", "character",
    "check", "checkpoint", "close", "comment",
    "commit", "connect", "constraint", "contains",
    "continue", "convert", "create", "cross",
    "cube", "current", "current_timestamp", "current_user",
    "cursor", "date", "dbspace", "deallocate",
    "dec", "decimal", "declare", "default",
    "delete", "deleting", "desc", "distinct",
    "do", "double", "drop", "dynamic",
    "else", "elseif", "encrypted", "end",
    "endif", "escape", "except", "exception",
    "exec", "execute", "existing", "exists",
    "externlogin", "fetch", "first", "float",
    "for", "force", "foreign", "forward",
    "from", "full", "goto", "grant",
    "group", "having", "holdlock", "identified",
    "if", "in", "index", "index_lparen",
    "inner", "inout", "insensitive", "insert",
    "inserting", "install", "instead", "int",
    "integer", "integrated", "intersect", "into",
    "iq", "is", "isolation", "join",
    "key", "lateral", "left", "like",
    "lock", "login", "long", "match",
    "membership", "message", "mode", "modify",
    "natural", "new", "no", "noholdlock",
    "not", "notify", "null", "numeric",
    "of", "off", "on", "open",
    "option", "options", "or", "order",
    "others", "out", "outer", "over",
    "passthrough", "precision", "prepare", "primary",
    "print", "privileges", "proc", "procedure",
    "publication", "raiserror", "readtext", "real",
    "reference", "references", "release", "remote",
    "remove", "rename", "reorganize", "resource",
    "restore", "restrict", "return", "revoke",
    "right", "rollback", "rollup", "save",
    "savepoint", "scroll", "select", "sensitive",
    "session", "set", "setuser", "share",
    "smallint", "some", "sqlcode", "sqlstate",
    "start", "stop", "subtrans", "subtransaction",
    "synchronize", "syntax_error", "table", "temporary",
    "then", "time", "timestamp", "tinyint",
    "to", "top", "tran", "trigger",
    "truncate", "tsequal", "unbounded", "union",
    "unique", "unknown", "unsigned", "update",
    "updating", "user", "using", "validate",
    "values", "varbinary", "varchar", "variable",
    "varying", "view", "wait", "waitfor",
    "when", "where", "while", "window",
    "with", "with_cube", "with_lparen", "with_rollup",
    "within", "work", "writetext",
    ])

ischema = MetaData()

tables = Table("SYSTABLE", ischema,
    Column("table_id", Integer, primary_key=True),
    Column("file_id", SMALLINT),
    Column("table_name", CHAR(128)),
    Column("table_type", CHAR(10)),
    Column("creator", Integer),
    #schema="information_schema"
    )

domains = Table("SYSDOMAIN", ischema,
    Column("domain_id", Integer, primary_key=True),
    Column("domain_name", CHAR(128)),
    Column("type_id", SMALLINT),
    Column("precision", SMALLINT, quote=True),
    #schema="information_schema"
    )

columns = Table("SYSCOLUMN", ischema,
    Column("column_id", Integer, primary_key=True),
    Column("table_id", Integer, ForeignKey(tables.c.table_id)),
    Column("pkey", CHAR(1)),
    Column("column_name", CHAR(128)),
    Column("nulls", CHAR(1)),
    Column("width", SMALLINT),
    Column("domain_id", SMALLINT, ForeignKey(domains.c.domain_id)),
    # FIXME: should be mx.BIGINT
    Column("max_identity", Integer),
    # FIXME: should be mx.ODBC.Windows.LONGVARCHAR
    Column("default", String),
    Column("scale", Integer),
    #schema="information_schema"
    )

foreignkeys = Table("SYSFOREIGNKEY", ischema,
    Column("foreign_table_id", Integer, ForeignKey(tables.c.table_id), primary_key=True),
    Column("foreign_key_id", SMALLINT, primary_key=True),
    Column("primary_table_id", Integer, ForeignKey(tables.c.table_id)),
    #schema="information_schema"
    )
fkcols = Table("SYSFKCOL", ischema,
    Column("foreign_table_id", Integer, ForeignKey(columns.c.table_id), primary_key=True),
    Column("foreign_key_id", SMALLINT, ForeignKey(foreignkeys.c.foreign_key_id), primary_key=True),
    Column("foreign_column_id", Integer, ForeignKey(columns.c.column_id), primary_key=True),
    Column("primary_column_id", Integer),
    #schema="information_schema"
    )

class SybaseTypeError(sqltypes.TypeEngine):
    def result_processor(self, dialect):
        return None

    def bind_processor(self, dialect):
        def process(value):
            raise exceptions.NotSupportedError("Data type not supported", [value])
        return process

    def get_col_spec(self):
        raise exceptions.NotSupportedError("Data type not supported")

class SybaseNumeric(sqltypes.Numeric):
    def get_col_spec(self):
        if self.length is None:
            if self.precision is None:
                return "NUMERIC"
            else:
                return "NUMERIC(%(precision)s)" % {'precision' : self.precision}
        else:
            return "NUMERIC(%(precision)s, %(length)s)" % {'precision': self.precision, 'length' : self.length}

class SybaseFloat(sqltypes.FLOAT, SybaseNumeric):
    def __init__(self, precision = 10, asdecimal = False, length = 2, **kwargs):
        super(sqltypes.FLOAT, self).__init__(precision, asdecimal, **kwargs)
        self.length = length

    def get_col_spec(self):
        # if asdecimal is True, handle same way as SybaseNumeric
        if self.asdecimal:
            return SybaseNumeric.get_col_spec(self)
        if self.precision is None:
            return "FLOAT"
        else:
            return "FLOAT(%(precision)s)" % {'precision': self.precision}

    def result_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            return float(value)
        if self.asdecimal:
            return SybaseNumeric.result_processor(self, dialect)
        return process

class SybaseInteger(sqltypes.Integer):
    def get_col_spec(self):
        return "INTEGER"

class SybaseBigInteger(SybaseInteger):
    def get_col_spec(self):
        return "BIGINT"

class SybaseTinyInteger(SybaseInteger):
    def get_col_spec(self):
        return "TINYINT"

class SybaseSmallInteger(SybaseInteger):
    def get_col_spec(self):
        return "SMALLINT"

class SybaseDateTime_mxodbc(sqltypes.DateTime):
    def __init__(self, *a, **kw):
        super(SybaseDateTime_mxodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

class SybaseDateTime_pyodbc(sqltypes.DateTime):
    def __init__(self, *a, **kw):
        super(SybaseDateTime_pyodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

    def result_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            # Convert the datetime.datetime back to datetime.time
            return value
        return process

    def bind_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            return value
        return process

class SybaseDate_mxodbc(sqltypes.Date):
    def __init__(self, *a, **kw):
        super(SybaseDate_mxodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATE"

class SybaseDate_pyodbc(sqltypes.Date):
    def __init__(self, *a, **kw):
        super(SybaseDate_pyodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATE"

class SybaseTime_mxodbc(sqltypes.Time):
    def __init__(self, *a, **kw):
        super(SybaseTime_mxodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

    def result_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            # Convert the datetime.datetime back to datetime.time
            return datetime.time(value.hour, value.minute, value.second, value.microsecond)
        return process

class SybaseTime_pyodbc(sqltypes.Time):
    def __init__(self, *a, **kw):
        super(SybaseTime_pyodbc, self).__init__(False)

    def get_col_spec(self):
        return "DATETIME"

    def result_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            # Convert the datetime.datetime back to datetime.time
            return datetime.time(value.hour, value.minute, value.second, value.microsecond)
        return process

    def bind_processor(self, dialect):
        def process(value):
            if value is None: