maxdb.py

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

    'integer': MaxInteger,
    'long binary': MaxBlob,
    'long unicode': MaxText,
    'long': MaxText,
    'long': MaxText,
    'smallint': MaxSmallInteger,
    'time': MaxTime,
    'timestamp': MaxTimestamp,
    'varchar': MaxString,
    }


class MaxDBExecutionContext(default.DefaultExecutionContext):
    def post_exec(self):
        # DB-API bug: if there were any functions as values,
        # then do another select and pull CURRVAL from the
        # autoincrement column's implicit sequence... ugh
        if self.compiled.isinsert and not self.executemany:
            table = self.compiled.statement.table
            index, serial_col = _autoserial_column(table)

            if serial_col and (not self.compiled._safeserial or
                               not(self._last_inserted_ids) or
                               self._last_inserted_ids[index] in (None, 0)):
                if table.schema:
                    sql = "SELECT %s.CURRVAL FROM DUAL" % (
                        self.compiled.preparer.format_table(table))
                else:
                    sql = "SELECT CURRENT_SCHEMA.%s.CURRVAL FROM DUAL" % (
                        self.compiled.preparer.format_table(table))

                if self.connection.engine._should_log_info:
                    self.connection.engine.logger.info(sql)
                rs = self.cursor.execute(sql)
                id = rs.fetchone()[0]

                if self.connection.engine._should_log_debug:
                    self.connection.engine.logger.debug([id])
                if not self._last_inserted_ids:
                    # This shouldn't ever be > 1?  Right?
                    self._last_inserted_ids = \
                      [None] * len(table.primary_key.columns)
                self._last_inserted_ids[index] = id

        super(MaxDBExecutionContext, self).post_exec()

    def get_result_proxy(self):
        if self.cursor.description is not None:
            for column in self.cursor.description:
                if column[1] in ('Long Binary', 'Long', 'Long Unicode'):
                    return MaxDBResultProxy(self)
        return engine_base.ResultProxy(self)


class MaxDBCachedColumnRow(engine_base.RowProxy):
    """A RowProxy that only runs result_processors once per column."""

    def __init__(self, parent, row):
        super(MaxDBCachedColumnRow, self).__init__(parent, row)
        self.columns = {}
        self._row = row
        self._parent = parent

    def _get_col(self, key):
        if key not in self.columns:
            self.columns[key] = self._parent._get_col(self._row, key)
        return self.columns[key]

    def __iter__(self):
        for i in xrange(len(self._row)):
            yield self._get_col(i)

    def __repr__(self):
        return repr(list(self))

    def __eq__(self, other):
        return ((other is self) or
                (other == tuple([self._get_col(key)
                                 for key in xrange(len(self._row))])))
    def __getitem__(self, key):
        if isinstance(key, slice):
            indices = key.indices(len(self._row))
            return tuple([self._get_col(i) for i in xrange(*indices)])
        else:
            return self._get_col(key)

    def __getattr__(self, name):
        try:
            return self._get_col(name)
        except KeyError:
            raise AttributeError(name)


class MaxDBResultProxy(engine_base.ResultProxy):
    _process_row = MaxDBCachedColumnRow


class MaxDBDialect(default.DefaultDialect):
    supports_alter = True
    supports_unicode_statements = True
    max_identifier_length = 32
    supports_sane_rowcount = True
    supports_sane_multi_rowcount = False
    preexecute_sequences = True

    # MaxDB-specific
    datetimeformat = 'internal'

    def __init__(self, _raise_known_sql_errors=False, **kw):
        super(MaxDBDialect, self).__init__(**kw)
        self._raise_known = _raise_known_sql_errors

        if self.dbapi is None:
            self.dbapi_type_map = {}
        else:
            self.dbapi_type_map = {
                'Long Binary': MaxBlob(),
                'Long byte_t': MaxBlob(),
                'Long Unicode': MaxText(),
                'Timestamp': MaxTimestamp(),
                'Date': MaxDate(),
                'Time': MaxTime(),
                datetime.datetime: MaxTimestamp(),
                datetime.date: MaxDate(),
                datetime.time: MaxTime(),
            }

    def dbapi(cls):
        from sapdb import dbapi as _dbapi
        return _dbapi
    dbapi = classmethod(dbapi)

    def create_connect_args(self, url):
        opts = url.translate_connect_args(username='user')
        opts.update(url.query)
        return [], opts

    def type_descriptor(self, typeobj):
        if isinstance(typeobj, type):
            typeobj = typeobj()
        if isinstance(typeobj, sqltypes.Unicode):
            return typeobj.adapt(MaxUnicode)
        else:
            return sqltypes.adapt_type(typeobj, colspecs)

    def create_execution_context(self, connection, **kw):
        return MaxDBExecutionContext(self, connection, **kw)

    def do_execute(self, cursor, statement, parameters, context=None):
        res = cursor.execute(statement, parameters)
        if isinstance(res, int) and context is not None:
            context._rowcount = res

    def do_release_savepoint(self, connection, name):
        # Does MaxDB truly support RELEASE SAVEPOINT <id>?  All my attempts
        # produce "SUBTRANS COMMIT/ROLLBACK not allowed without SUBTRANS
        # BEGIN SQLSTATE: I7065"
        # Note that ROLLBACK TO works fine.  In theory, a RELEASE should
        # just free up some transactional resources early, before the overall
        # COMMIT/ROLLBACK so omitting it should be relatively ok.
        pass

    def get_default_schema_name(self, connection):
        try:
            return self._default_schema_name
        except AttributeError:
            name = self.identifier_preparer._normalize_name(
                connection.execute('SELECT CURRENT_SCHEMA FROM DUAL').scalar())
            self._default_schema_name = name
            return name

    def has_table(self, connection, table_name, schema=None):
        denormalize = self.identifier_preparer._denormalize_name
        bind = [denormalize(table_name)]
        if schema is None:
            sql = ("SELECT tablename FROM TABLES "
                   "WHERE TABLES.TABLENAME=? AND"
                   "  TABLES.SCHEMANAME=CURRENT_SCHEMA ")
        else:
            sql = ("SELECT tablename FROM TABLES "
                   "WHERE TABLES.TABLENAME = ? AND"
                   "  TABLES.SCHEMANAME=? ")
            bind.append(denormalize(schema))

        rp = connection.execute(sql, bind)
        found = bool(rp.fetchone())
        rp.close()
        return found

    def table_names(self, connection, schema):
        if schema is None:
            sql = (" SELECT TABLENAME FROM TABLES WHERE "
                   " SCHEMANAME=CURRENT_SCHEMA ")
            rs = connection.execute(sql)
        else:
            sql = (" SELECT TABLENAME FROM TABLES WHERE "
                   " SCHEMANAME=? ")
            matchname = self.identifier_preparer._denormalize_name(schema)
            rs = connection.execute(sql, matchname)
        normalize = self.identifier_preparer._normalize_name
        return [normalize(row[0]) for row in rs]

    def reflecttable(self, connection, table, include_columns):
        denormalize = self.identifier_preparer._denormalize_name
        normalize = self.identifier_preparer._normalize_name

        st = ('SELECT COLUMNNAME, MODE, DATATYPE, CODETYPE, LEN, DEC, '
              '  NULLABLE, "DEFAULT", DEFAULTFUNCTION '
              'FROM COLUMNS '
              'WHERE TABLENAME=? AND SCHEMANAME=%s '
              'ORDER BY POS')

        fk = ('SELECT COLUMNNAME, FKEYNAME, '
              '  REFSCHEMANAME, REFTABLENAME, REFCOLUMNNAME, RULE, '
              '  (CASE WHEN REFSCHEMANAME = CURRENT_SCHEMA '
              '   THEN 1 ELSE 0 END) AS in_schema '
              'FROM FOREIGNKEYCOLUMNS '
              'WHERE TABLENAME=? AND SCHEMANAME=%s '
              'ORDER BY FKEYNAME ')

        params = [denormalize(table.name)]
        if not table.schema:
            st = st % 'CURRENT_SCHEMA'
            fk = fk % 'CURRENT_SCHEMA'
        else:
            st = st % '?'
            fk = fk % '?'
            params.append(denormalize(table.schema))

        rows = connection.execute(st, params).fetchall()
        if not rows:
            raise exceptions.NoSuchTableError(table.fullname)

        include_columns = util.Set(include_columns or [])

        for row in rows:
            (name, mode, col_type, encoding, length, scale,
             nullable, constant_def, func_def) = row

            name = normalize(name)

            if include_columns and name not in include_columns:
                continue

            type_args, type_kw = [], {}
            if col_type == 'FIXED':
                type_args = length, scale
                # Convert FIXED(10) DEFAULT SERIAL to our Integer
                if (scale == 0 and
                    func_def is not None and func_def.startswith('SERIAL')):
                    col_type = 'INTEGER'
                    type_args = length,
            elif col_type in 'FLOAT':
                type_args = length,
            elif col_type in ('CHAR', 'VARCHAR'):
                type_args = length,
                type_kw['encoding'] = encoding
            elif col_type == 'LONG':
                type_kw['encoding'] = encoding

            try:
                type_cls = ischema_names[col_type.lower()]
                type_instance = type_cls(*type_args, **type_kw)
            except KeyError:
                util.warn("Did not recognize type '%s' of column '%s'" %
                          (col_type, name))
                type_instance = sqltypes.NullType

            col_kw = {'autoincrement': False}
            col_kw['nullable'] = (nullable == 'YES')
            col_kw['primary_key'] = (mode == 'KEY')

            if func_def is not None:
                if func_def.startswith('SERIAL'):
                    if col_kw['primary_key']:
                        # No special default- let the standard autoincrement
                        # support handle SERIAL pk columns.
                        col_kw['autoincrement'] = True
                    else:
                        # strip current numbering
                        col_kw['default'] = schema.PassiveDefault(
                            sql.text('SERIAL'))
                        col_kw['autoincrement'] = True
                else:
                    col_kw['default'] = schema.PassiveDefault(
                        sql.text(func_def))
            elif constant_def is not None:
                col_kw['default'] = schema.PassiveDefault(sql.text(
                    "'%s'" % constant_def.replace("'", "''")))

            table.append_column(schema.Column(name, type_instance, **col_kw))

        fk_sets = itertools.groupby(connection.execute(fk, params),
                                    lambda row: row.FKEYNAME)
        for fkeyname, fkey in fk_sets:
            fkey = list(fkey)
            if include_columns:
                key_cols = util.Set([r.COLUMNNAME for r in fkey])
                if key_cols != include_columns:
                    continue

            columns, referants = [], []
            quote = self.identifier_preparer._maybe_quote_identifier

            for row in fkey:
                columns.append(normalize(row.COLUMNNAME))
                if table.schema or not row.in_schema:
                    referants.append('.'.join(
                        [quote(normalize(row[c]))
                         for c in ('REFSCHEMANAME', 'REFTABLENAME',
                                   'REFCOLUMNNAME')]))
                else:
                    referants.append('.'.join(
                        [quote(normalize(row[c]))
                         for c in ('REFTABLENAME', 'REFCOLUMNNAME')]))

            constraint_kw = {'name': fkeyname.lower()}
            if fkey[0].RULE is not None:
                rule = fkey[0].RULE
                if rule.startswith('DELETE '):
                    rule = rule[7:]
                constraint_kw['ondelete'] = rule

            table_kw = {}
            if table.schema or not row.in_schema:
                table_kw['schema'] = normalize(fkey[0].REFSCHEMANAME)

            ref_key = schema._get_table_key(normalize(fkey[0].REFTABLENAME),
                                            table_kw.get('schema'))
            if ref_key not in table.metadata.tables:
                schema.Table(normalize(fkey[0].REFTABLENAME),
                             table.metadata,
                             autoload=True, autoload_with=connection,
                             **table_kw)

            constraint = schema.ForeignKeyConstraint(columns, referants,
                                                     **constraint_kw)
            table.append_constraint(constraint)

    def has_sequence(self, connection, name):
        # [ticket:726] makes this schema-aware.
        denormalize = self.identifier_preparer._denormalize_name
        sql = ("SELECT sequence_name FROM SEQUENCES "
               "WHERE SEQUENCE_NAME=? ")

        rp = connection.execute(sql, denormalize(name))
        found = bool(rp.fetchone())
        rp.close()
        return found


class MaxDBCompiler(compiler.DefaultCompiler):
    operators = compiler.DefaultCompiler.operators.copy()
    operators[sql_operators.mod] = lambda x, y: 'mod(%s, %s)' % (x, y)

    function_conversion = {
        'CURRENT_DATE': 'DATE',
        'CURRENT_TIME': 'TIME',
        'CURRENT_TIMESTAMP': 'TIMESTAMP',
        }

    # These functions must be written without parens when called with no
    # parameters.  e.g. 'SELECT DATE FROM DUAL' not 'SELECT DATE() FROM DUAL'
    bare_functions = util.Set([
        'CURRENT_SCHEMA', 'DATE', 'FALSE', 'SYSDBA', 'TIME', 'TIMESTAMP',
        'TIMEZONE', 'TRANSACTION', 'TRUE', 'USER', 'UID', 'USERGROUP',
        'UTCDATE', 'UTCDIFF'])

    def default_from(self):
        return ' FROM DUAL'