oracle.py

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

# oracle.py
# Copyright (C) 2005, 2006, 2007, 2008 Michael Bayer [e-mail]
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php


import datetime, random, re

from sqlalchemy import util, sql, schema, exceptions, logging
from sqlalchemy.engine import default, base
from sqlalchemy.sql import compiler, visitors
from sqlalchemy.sql import operators as sql_operators, functions as sql_functions
from sqlalchemy import types as sqltypes


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

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

class OracleSmallInteger(sqltypes.Smallinteger):
    def get_col_spec(self):
        return "SMALLINT"

class OracleDate(sqltypes.Date):
    def get_col_spec(self):
        return "DATE"
    def bind_processor(self, dialect):
        return None

    def result_processor(self, dialect):
        def process(value):
            if not isinstance(value, datetime.datetime):
                return value
            else:
                return value.date()
        return process

class OracleDateTime(sqltypes.DateTime):
    def get_col_spec(self):
        return "DATE"

    def result_processor(self, dialect):
        def process(value):
            if value is None or isinstance(value,datetime.datetime):
                return value
            else:
                # convert cx_oracle datetime object returned pre-python 2.4
                return datetime.datetime(value.year,value.month,
                    value.day,value.hour, value.minute, value.second)
        return process

# Note:
# Oracle DATE == DATETIME
# Oracle does not allow milliseconds in DATE
# Oracle does not support TIME columns

# only if cx_oracle contains TIMESTAMP
class OracleTimestamp(sqltypes.TIMESTAMP):
    def get_col_spec(self):
        return "TIMESTAMP"

    def get_dbapi_type(self, dialect):
        return dialect.TIMESTAMP

    def result_processor(self, dialect):
        def process(value):
            if value is None or isinstance(value,datetime.datetime):
                return value
            else:
                # convert cx_oracle datetime object returned pre-python 2.4
                return datetime.datetime(value.year,value.month,
                    value.day,value.hour, value.minute, value.second)
        return process

class OracleString(sqltypes.String):
    def get_col_spec(self):
        return "VARCHAR(%(length)s)" % {'length' : self.length}

class OracleText(sqltypes.Text):
    def get_dbapi_type(self, dbapi):
        return dbapi.CLOB

    def get_col_spec(self):
        return "CLOB"

    def result_processor(self, dialect):
        super_process = super(OracleText, self).result_processor(dialect)
        lob = dialect.dbapi.LOB
        def process(value):
            if isinstance(value, lob):
                if super_process:
                    return super_process(value.read())
                else:
                    return value.read()
            else:
                if super_process:
                    return super_process(value)
                else:
                    return value
        return process

class OracleRaw(sqltypes.Binary):
    def get_col_spec(self):
        return "RAW(%(length)s)" % {'length' : self.length}

class OracleChar(sqltypes.CHAR):
    def get_col_spec(self):
        return "CHAR(%(length)s)" % {'length' : self.length}

class OracleBinary(sqltypes.Binary):
    def get_dbapi_type(self, dbapi):
        return dbapi.BLOB

    def get_col_spec(self):
        return "BLOB"

    def bind_processor(self, dialect):
        return None

    def result_processor(self, dialect):
        lob = dialect.dbapi.LOB
        def process(value):
            if isinstance(value, lob):
                return value.read()
            else:
                return value
        return process

class OracleBoolean(sqltypes.Boolean):
    def get_col_spec(self):
        return "SMALLINT"

    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

colspecs = {
    sqltypes.Integer : OracleInteger,
    sqltypes.Smallinteger : OracleSmallInteger,
    sqltypes.Numeric : OracleNumeric,
    sqltypes.Float : OracleNumeric,
    sqltypes.DateTime : OracleDateTime,
    sqltypes.Date : OracleDate,
    sqltypes.String : OracleString,
    sqltypes.Binary : OracleBinary,
    sqltypes.Boolean : OracleBoolean,
    sqltypes.Text : OracleText,
    sqltypes.TIMESTAMP : OracleTimestamp,
    sqltypes.CHAR: OracleChar,
}

ischema_names = {
    'VARCHAR2' : OracleString,
    'DATE' : OracleDateTime,
    'DATETIME' : OracleDateTime,
    'NUMBER' : OracleNumeric,
    'BLOB' : OracleBinary,
    'CLOB' : OracleText,
    'TIMESTAMP' : OracleTimestamp,
    'RAW' : OracleRaw,
    'FLOAT' : OracleNumeric,
    'DOUBLE PRECISION' : OracleNumeric,
    'LONG' : OracleText,
}

def descriptor():
    return {'name':'oracle',
    'description':'Oracle',
    'arguments':[
        ('dsn', 'Data Source Name', None),
        ('user', 'Username', None),
        ('password', 'Password', None)
    ]}

class OracleExecutionContext(default.DefaultExecutionContext):
    def pre_exec(self):
        super(OracleExecutionContext, self).pre_exec()
        if self.dialect.auto_setinputsizes:
            self.set_input_sizes()
        if self.compiled_parameters is not None and len(self.compiled_parameters) == 1:
            for key in self.compiled.binds:
                bindparam = self.compiled.binds[key]
                name = self.compiled.bind_names[bindparam]
                value = self.compiled_parameters[0][name]
                if bindparam.isoutparam:
                    dbtype = bindparam.type.dialect_impl(self.dialect).get_dbapi_type(self.dialect.dbapi)
                    if not hasattr(self, 'out_parameters'):
                        self.out_parameters = {}
                    self.out_parameters[name] = self.cursor.var(dbtype)
                    self.parameters[0][name] = self.out_parameters[name]

    def get_result_proxy(self):
        if hasattr(self, 'out_parameters'):
            if self.compiled_parameters is not None and len(self.compiled_parameters) == 1:
                 for bind, name in self.compiled.bind_names.iteritems():
                     if name in self.out_parameters:
                         type = bind.type
                         self.out_parameters[name] = type.dialect_impl(self.dialect).result_processor(self.dialect)(self.out_parameters[name].getvalue())
            else:
                 for k in self.out_parameters:
                     self.out_parameters[k] = self.out_parameters[k].getvalue()

        if self.cursor.description is not None:
            for column in self.cursor.description:
                type_code = column[1]
                if type_code in self.dialect.ORACLE_BINARY_TYPES:
                    return base.BufferedColumnResultProxy(self)

        return base.ResultProxy(self)

class OracleDialect(default.DefaultDialect):
    supports_alter = True
    supports_unicode_statements = False
    max_identifier_length = 30
    supports_sane_rowcount = True
    supports_sane_multi_rowcount = False
    preexecute_pk_sequences = True
    supports_pk_autoincrement = False

    def __init__(self, use_ansi=True, auto_setinputsizes=True, auto_convert_lobs=True, threaded=True, allow_twophase=True, **kwargs):
        default.DefaultDialect.__init__(self, default_paramstyle='named', **kwargs)
        self.use_ansi = use_ansi
        self.threaded = threaded
        self.allow_twophase = allow_twophase
        self.supports_timestamp = self.dbapi is None or hasattr(self.dbapi, 'TIMESTAMP' )
        self.auto_setinputsizes = auto_setinputsizes
        self.auto_convert_lobs = auto_convert_lobs
        if self.dbapi is None or not self.auto_convert_lobs or not 'CLOB' in self.dbapi.__dict__:
            self.dbapi_type_map = {}
            self.ORACLE_BINARY_TYPES = []
        else:
            # only use this for LOB objects.  using it for strings, dates
            # etc. leads to a little too much magic, reflection doesn't know if it should
            # expect encoded strings or unicodes, etc.
            self.dbapi_type_map = {
                self.dbapi.CLOB: OracleText(),
                self.dbapi.BLOB: OracleBinary(),
                self.dbapi.BINARY: OracleRaw(),
            }
            self.ORACLE_BINARY_TYPES = [getattr(self.dbapi, k) for k in ["BFILE", "CLOB", "NCLOB", "BLOB"] if hasattr(self.dbapi, k)]

    def dbapi(cls):
        import cx_Oracle
        return cx_Oracle
    dbapi = classmethod(dbapi)

    def create_connect_args(self, url):
        dialect_opts = dict(url.query)
        for opt in ('use_ansi', 'auto_setinputsizes', 'auto_convert_lobs',
                    'threaded', 'allow_twophase'):
            if opt in dialect_opts:
                util.coerce_kw_type(dialect_opts, opt, bool)
                setattr(self, opt, dialect_opts[opt])

        if url.database:
            # if we have a database, then we have a remote host
            port = url.port
            if port:
                port = int(port)
            else:
                port = 1521
            dsn = self.dbapi.makedsn(url.host, port, url.database)
        else:
            # we have a local tnsname
            dsn = url.host

        opts = dict(
            user=url.username,
            password=url.password,
            dsn=dsn,
            threaded=self.threaded,
            twophase=self.allow_twophase,
            )
        if 'mode' in url.query:
            opts['mode'] = url.query['mode']
            if isinstance(opts['mode'], basestring):
                mode = opts['mode'].upper()
                if mode == 'SYSDBA':
                    opts['mode'] = self.dbapi.SYSDBA
                elif mode == 'SYSOPER':
                    opts['mode'] = self.dbapi.SYSOPER
                else:
                    util.coerce_kw_type(opts, 'mode', int)
        # Can't set 'handle' or 'pool' via URL query args, use connect_args

        return ([], opts)

    def is_disconnect(self, e):
        if isinstance(e, self.dbapi.InterfaceError):
            return "not connected" in str(e)
        else:
            return "ORA-03114" in str(e) or "ORA-03113" in str(e)

    def type_descriptor(self, typeobj):
        return sqltypes.adapt_type(typeobj, colspecs)

    def oid_column_name(self, column):
        if not isinstance(column.table, (sql.TableClause, sql.Select)):
            return None
        else:
            return "rowid"

    def create_xid(self):
        """create a two-phase transaction ID.

        this id will be passed to do_begin_twophase(), do_rollback_twophase(),
        do_commit_twophase().  its format is unspecified."""

        id = random.randint(0,2**128)
        return (0x1234, "%032x" % 9, "%032x" % id)

    def do_release_savepoint(self, connection, name):
        # Oracle does not support RELEASE SAVEPOINT
        pass

    def do_begin_twophase(self, connection, xid):
        connection.connection.begin(*xid)

    def do_prepare_twophase(self, connection, xid):
        connection.connection.prepare()

    def do_rollback_twophase(self, connection, xid, is_prepared=True, recover=False):
        self.do_rollback(connection.connection)

    def do_commit_twophase(self, connection, xid, is_prepared=True, recover=False):
        self.do_commit(connection.connection)

    def do_recover_twophase(self, connection):
        pass

    def create_execution_context(self, *args, **kwargs):
        return OracleExecutionContext(self, *args, **kwargs)

    def has_table(self, connection, table_name, schema=None):
        cursor = connection.execute("""select table_name from all_tables where table_name=:name""", {'name':self._denormalize_name(table_name)})
        return bool( cursor.fetchone() is not None )

    def has_sequence(self, connection, sequence_name):
        cursor = connection.execute("""select sequence_name from all_sequences where sequence_name=:name""", {'name':self._denormalize_name(sequence_name)})
        return bool( cursor.fetchone() is not None )

    def _locate_owner_row(self, owner, name, rows, raiseerr=False):
        """return the row in the given list of rows which references the given table name and owner name."""
        if not rows:
            if raiseerr:
                raise exceptions.NoSuchTableError(name)