types.py

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

# types.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

"""defines genericized SQL types, each represented by a subclass of
[sqlalchemy.types#AbstractType].  Dialects define further subclasses of these
types.

For more information see the SQLAlchemy documentation on types.

"""
__all__ = [ 'TypeEngine', 'TypeDecorator', 'AbstractType',
            'INT', 'CHAR', 'VARCHAR', 'NCHAR', 'TEXT', 'Text', 'FLOAT',
            'NUMERIC', 'DECIMAL', 'TIMESTAMP', 'DATETIME', 'CLOB', 'BLOB',
            'BOOLEAN', 'SMALLINT', 'DATE', 'TIME',
            'String', 'Integer', 'SmallInteger','Smallinteger',
            'Numeric', 'Float', 'DateTime', 'Date', 'Time', 'Binary',
            'Boolean', 'Unicode', 'UnicodeText', 'PickleType', 'Interval',
            'type_map'
            ]

import inspect
import datetime as dt

from sqlalchemy import exceptions
from sqlalchemy.util import pickle, Decimal as _python_Decimal
import sqlalchemy.util as util
NoneType = type(None)

class _UserTypeAdapter(type):
    """adapts 0.3 style user-defined types with convert_bind_param/convert_result_value
    to use newer bind_processor()/result_processor() methods."""

    def __init__(cls, clsname, bases, dict):
        if not hasattr(cls.convert_result_value, '_sa_override'):
            cls.__instrument_result_proc(cls)

        if not hasattr(cls.convert_bind_param, '_sa_override'):
            cls.__instrument_bind_proc(cls)

        return super(_UserTypeAdapter, cls).__init__(clsname, bases, dict)

    def __instrument_bind_proc(cls, class_):
        def bind_processor(self, dialect):
            def process(value):
                return self.convert_bind_param(value, dialect)
            return process
        class_.super_bind_processor = class_.bind_processor
        class_.bind_processor = bind_processor

    def __instrument_result_proc(cls, class_):
        def result_processor(self, dialect):
            def process(value):
                return self.convert_result_value(value, dialect)
            return process
        class_.super_result_processor = class_.result_processor
        class_.result_processor = result_processor


class AbstractType(object):
    __metaclass__ = _UserTypeAdapter

    def __init__(self, *args, **kwargs):
        pass

    def copy_value(self, value):
        return value

    def convert_result_value(self, value, dialect):
        """Legacy convert_result_value() compatibility method.

        This adapter method is provided for user-defined types that implement
        the older convert_* interface and need to call their super method.
        These calls are adapted behind the scenes to use the newer
        callable-based interface via result_processor().

        Compatibility is configured on a case-by-case basis at class
        definition time by a legacy adapter metaclass.  This method is only
        available and functional if the concrete subclass implements the
        legacy interface.
        """

        processor = self.super_result_processor(dialect)
        if processor:
            return processor(value)
        else:
            return value
    convert_result_value._sa_override = True

    def convert_bind_param(self, value, dialect):
        """Legacy convert_bind_param() compatability method.

        This adapter method is provided for user-defined types that implement
        the older convert_* interface and need to call their super method.
        These calls are adapted behind the scenes to use the newer
        callable-based interface via bind_processor().

        Compatibility is configured on a case-by-case basis at class
        definition time by a legacy adapter metaclass.  This method is only
        available and functional if the concrete subclass implements the
        legacy interface.
        """

        processor = self.super_bind_processor(dialect)
        if processor:
            return processor(value)
        else:
            return value
    convert_bind_param._sa_override = True

    def bind_processor(self, dialect):
        """Defines a bind parameter processing function."""

        return None

    def result_processor(self, dialect):
        """Defines a result-column processing function."""

        return None

    def compare_values(self, x, y):
        """compare two values for equality."""

        return x == y

    def is_mutable(self):
        """return True if the target Python type is 'mutable'.

        This allows systems like the ORM to know if an object
        can be considered 'not changed' by identity alone.
        """

        return False

    def get_dbapi_type(self, dbapi):
        """Return the corresponding type object from the underlying DB-API, if any.

        This can be useful for calling ``setinputsizes()``, for example.
        """

        return None

    def adapt_operator(self, op):
        """given an operator from the sqlalchemy.sql.operators package,
        translate it to a new operator based on the semantics of this type.

        By default, returns the operator unchanged."""
        return op

    def __repr__(self):
        return "%s(%s)" % (
            self.__class__.__name__,
            ", ".join(["%s=%r" % (k, getattr(self, k, None))
                      for k in inspect.getargspec(self.__init__)[0][1:]]))

class TypeEngine(AbstractType):
    def dialect_impl(self, dialect, **kwargs):
        try:
            return self._impl_dict[dialect]
        except AttributeError:
            self._impl_dict = {}
            return self._impl_dict.setdefault(dialect, dialect.type_descriptor(self))
        except KeyError:
            return self._impl_dict.setdefault(dialect, dialect.type_descriptor(self))

    def __getstate__(self):
        d = self.__dict__.copy()
        d['_impl_dict'] = {}
        return d

    def get_col_spec(self):
        raise NotImplementedError()


    def bind_processor(self, dialect):
        return None

    def result_processor(self, dialect):
        return None

    def adapt(self, cls):
        return cls()

    def get_search_list(self):
        """return a list of classes to test for a match
        when adapting this type to a dialect-specific type.

        """

        return self.__class__.__mro__[0:-1]

class TypeDecorator(AbstractType):
    def __init__(self, *args, **kwargs):
        if not hasattr(self.__class__, 'impl'):
            raise exceptions.AssertionError("TypeDecorator implementations require a class-level variable 'impl' which refers to the class of type being decorated")
        self.impl = self.__class__.impl(*args, **kwargs)

    def dialect_impl(self, dialect, **kwargs):
        try:
            return self._impl_dict[dialect]
        except AttributeError:
            self._impl_dict = {}
        except KeyError:
            pass

        if isinstance(self.impl, TypeDecorator):
            typedesc = self.impl.dialect_impl(dialect)
        else:
            typedesc = self.load_dialect_impl(dialect)
        tt = self.copy()
        if not isinstance(tt, self.__class__):
            raise exceptions.AssertionError("Type object %s does not properly implement the copy() method, it must return an object of type %s" % (self, self.__class__))
        tt.impl = typedesc
        self._impl_dict[dialect] = tt
        return tt

    def load_dialect_impl(self, dialect):
        """loads the dialect-specific implementation of this type.

        by default calls dialect.type_descriptor(self.impl), but
        can be overridden to provide different behavior.
        """

        return dialect.type_descriptor(self.impl)

    def __getattr__(self, key):
        """Proxy all other undefined accessors to the underlying implementation."""

        return getattr(self.impl, key)

    def get_col_spec(self):
        return self.impl.get_col_spec()

    def process_bind_param(self, value, dialect):
        raise NotImplementedError()

    def process_result_value(self, value, dialect):
        raise NotImplementedError()

    def bind_processor(self, dialect):
        if self.__class__.process_bind_param.func_code is not TypeDecorator.process_bind_param.func_code:
            impl_processor = self.impl.bind_processor(dialect)
            if impl_processor:
                def process(value):
                    return impl_processor(self.process_bind_param(value, dialect))
                return process
            else:
                def process(value):
                    return self.process_bind_param(value, dialect)
                return process
        else:
            return self.impl.bind_processor(dialect)

    def result_processor(self, dialect):
        if self.__class__.process_result_value.func_code is not TypeDecorator.process_result_value.func_code:
            impl_processor = self.impl.result_processor(dialect)
            if impl_processor:
                def process(value):
                    return self.process_result_value(impl_processor(value), dialect)
                return process
            else:
                def process(value):
                    return self.process_result_value(value, dialect)
                return process
        else:
            return self.impl.result_processor(dialect)

    def copy(self):
        instance = self.__class__.__new__(self.__class__)
        instance.__dict__.update(self.__dict__)
        return instance

    def get_dbapi_type(self, dbapi):
        return self.impl.get_dbapi_type(dbapi)

    def copy_value(self, value):
        return self.impl.copy_value(value)

    def compare_values(self, x, y):
        return self.impl.compare_values(x,y)

    def is_mutable(self):
        return self.impl.is_mutable()

class MutableType(object):
    """A mixin that marks a Type as holding a mutable object."""

    def is_mutable(self):
        return True

    def copy_value(self, value):
        raise NotImplementedError()

    def compare_values(self, x, y):
        return x == y

def to_instance(typeobj):
    if typeobj is None:
        return NULLTYPE
    elif isinstance(typeobj, type):
        return typeobj()
    else:
        return typeobj

def adapt_type(typeobj, colspecs):
    if isinstance(typeobj, type):
        typeobj = typeobj()
    for t in typeobj.get_search_list():
        try:
            impltype = colspecs[t]
            break
        except KeyError:
            pass
    else:
        # couldnt adapt - so just return the type itself
        # (it may be a user-defined type)
        return typeobj
    # if we adapted the given generic type to a database-specific type,
    # but it turns out the originally given "generic" type
    # is actually a subclass of our resulting type, then we were already
    # given a more specific type than that required; so use that.
    if (issubclass(typeobj.__class__, impltype)):
        return typeobj
    return typeobj.adapt(impltype)

class NullType(TypeEngine):
    def get_col_spec(self):
        raise NotImplementedError()

NullTypeEngine = NullType

class Concatenable(object):
    """marks a type as supporting 'concatenation'"""
    def adapt_operator(self, op):
        from sqlalchemy.sql import operators
        if op == operators.add:
            return operators.concat_op
        else:
            return op

class String(Concatenable, TypeEngine):
    """A sized string type.

    Usually corresponds to VARCHAR.  Can also take Python unicode objects
    and encode to the database's encoding in bind params (and the reverse for
    result sets.)

    a String with no length will adapt itself automatically to a Text
    object at the dialect level (this behavior is deprecated in 0.4).
    """
    def __init__(self, length=None, convert_unicode=False, assert_unicode=None):