types.py

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

        self.length = length
        self.convert_unicode = convert_unicode
        self.assert_unicode = assert_unicode

    def adapt(self, impltype):
        return impltype(length=self.length, convert_unicode=self.convert_unicode, assert_unicode=self.assert_unicode)

    def bind_processor(self, dialect):
        if self.convert_unicode or dialect.convert_unicode:
            if self.assert_unicode is None:
                assert_unicode = dialect.assert_unicode
            else:
                assert_unicode = self.assert_unicode
            def process(value):
                if isinstance(value, unicode):
                    return value.encode(dialect.encoding)
                elif assert_unicode and not isinstance(value, (unicode, NoneType)):
                    if assert_unicode == 'warn':
                        util.warn("Unicode type received non-unicode bind "
                                  "param value %r" % value)
                        return value
                    else:
                        raise exceptions.InvalidRequestError("Unicode type received non-unicode bind param value %r" % value)
                else:
                    return value
            return process
        else:
            return None

    def result_processor(self, dialect):
        if self.convert_unicode or dialect.convert_unicode:
            def process(value):
                if value is not None and not isinstance(value, unicode):
                    return value.decode(dialect.encoding)
                else:
                    return value
            return process
        else:
            return None

    def dialect_impl(self, dialect, **kwargs):
        _for_ddl = kwargs.pop('_for_ddl', False)
        if _for_ddl and self.length is None:
            label = util.to_ascii(_for_ddl is True and
                                  '' or (' for column "%s"' % str(_for_ddl)))
            util.warn_deprecated(
                "Using String type with no length for CREATE TABLE "
                "is deprecated; use the Text type explicitly" + label)
        return TypeEngine.dialect_impl(self, dialect, **kwargs)

    def get_search_list(self):
        l = super(String, self).get_search_list()
        # if we are String or Unicode with no length,
        # return Text as the highest-priority type
        # to be adapted by the dialect
        if self.length is None and l[0] in (String, Unicode):
            return (Text,) + l
        else:
            return l

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

class Text(String):
    def dialect_impl(self, dialect, **kwargs):
        return TypeEngine.dialect_impl(self, dialect, **kwargs)

class Unicode(String):
    """A synonym for String(length, convert_unicode=True, assert_unicode='warn')."""

    def __init__(self, length=None, **kwargs):
        kwargs['convert_unicode'] = True
        kwargs['assert_unicode'] = 'warn'
        super(Unicode, self).__init__(length=length, **kwargs)

class UnicodeText(Text):
    """A synonym for Text(convert_unicode=True, assert_unicode='warn')."""

    def __init__(self, length=None, **kwargs):
        kwargs['convert_unicode'] = True
        kwargs['assert_unicode'] = 'warn'
        super(UnicodeText, self).__init__(length=length, **kwargs)

class Integer(TypeEngine):
    """Integer datatype."""

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

class SmallInteger(Integer):
    """Smallint datatype."""

Smallinteger = SmallInteger

class Numeric(TypeEngine):
    """Numeric datatype, usually resolves to DECIMAL or NUMERIC."""

    def __init__(self, precision=10, length=2, asdecimal=True):
        self.precision = precision
        self.length = length
        self.asdecimal = asdecimal

    def adapt(self, impltype):
        return impltype(precision=self.precision, length=self.length, asdecimal=self.asdecimal)

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

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

    def result_processor(self, dialect):
        if self.asdecimal:
            def process(value):
                if value is not None:
                    return _python_Decimal(str(value))
                else:
                    return value
            return process
        else:
            return None

class Float(Numeric):
    def __init__(self, precision = 10, asdecimal=False, **kwargs):
        self.precision = precision
        self.asdecimal = asdecimal

    def adapt(self, impltype):
        return impltype(precision=self.precision, asdecimal=self.asdecimal)

class DateTime(TypeEngine):
    """Implement a type for ``datetime.datetime()`` objects."""

    def __init__(self, timezone=False):
        self.timezone = timezone

    def adapt(self, impltype):
        return impltype(timezone=self.timezone)

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

class Date(TypeEngine):
    """Implement a type for ``datetime.date()`` objects."""

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

class Time(TypeEngine):
    """Implement a type for ``datetime.time()`` objects."""

    def __init__(self, timezone=False):
        self.timezone = timezone

    def adapt(self, impltype):
        return impltype(timezone=self.timezone)

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

class Binary(TypeEngine):
    def __init__(self, length=None):
        self.length = length

    def bind_processor(self, dialect):
        DBAPIBinary = dialect.dbapi.Binary
        def process(value):
            if value is not None:
                return DBAPIBinary(value)
            else:
                return None
        return process

    def adapt(self, impltype):
        return impltype(length=self.length)

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

class PickleType(MutableType, TypeDecorator):
    impl = Binary

    def __init__(self, protocol=pickle.HIGHEST_PROTOCOL, pickler=None, mutable=True, comparator=None):
        self.protocol = protocol
        self.pickler = pickler or pickle
        self.mutable = mutable
        self.comparator = comparator
        super(PickleType, self).__init__()

    def bind_processor(self, dialect):
        impl_process = self.impl.bind_processor(dialect)
        dumps = self.pickler.dumps
        protocol = self.protocol
        if impl_process is None:
            def process(value):
                if value is None:
                    return None
                return dumps(value, protocol)
        else:
            def process(value):
                if value is None:
                    return None
                return impl_process(dumps(value, protocol))
        return process

    def result_processor(self, dialect):
        impl_process = self.impl.result_processor(dialect)
        loads = self.pickler.loads
        if impl_process is None:
            def process(value):
                if value is None:
                    return None
                return loads(str(value))
        else:
            def process(value):
                if value is None:
                    return None
                return loads(str(impl_process(value)))
        return process

    def copy_value(self, value):
        if self.mutable:
            return self.pickler.loads(self.pickler.dumps(value, self.protocol))
        else:
            return value

    def compare_values(self, x, y):
        if self.comparator:
            return self.comparator(x, y)
        elif self.mutable:
            return self.pickler.dumps(x, self.protocol) == self.pickler.dumps(y, self.protocol)
        else:
            return x == y

    def is_mutable(self):
        return self.mutable

class Boolean(TypeEngine):
    pass

class Interval(TypeDecorator):
    """Type to be used in Column statements to store python timedeltas.

        If it's possible it uses native engine features to store timedeltas
        (now it's only PostgreSQL Interval type), if there is no such it
        fallbacks to DateTime storage with converting from/to timedelta on the fly

        Converting is very simple - just use epoch(zero timestamp, 01.01.1970) as
        base, so if we need to store timedelta = 1 day (24 hours) in database it
        will be stored as DateTime = '2nd Jan 1970 00:00', see bind_processor
        and result_processor to actual conversion code
    """
    #Empty useless type, because at the moment of creation of instance we don't
    #know what type will be decorated - it depends on used dialect.
    impl = TypeEngine

    def load_dialect_impl(self, dialect):
        """Checks if engine has native implementation of timedelta python type,
        if so it returns right class to handle it, if there is no native support,
        it fallback to engine's DateTime implementation class
        """
        if not hasattr(self,'__supported'):
            import sqlalchemy.databases.postgres as pg
            self.__supported = {pg.PGDialect:pg.PGInterval}
            del pg

        if self.__hasNativeImpl(dialect):
            #For now, only PostgreSQL has native timedelta types support
            return self.__supported[dialect.__class__]()
        else:
            #All others should fallback to DateTime
            return dialect.type_descriptor(DateTime)

    def __hasNativeImpl(self,dialect):
        return dialect.__class__ in self.__supported

    def bind_processor(self, dialect):
        impl_processor = self.impl.bind_processor(dialect)
        if self.__hasNativeImpl(dialect):
            return impl_processor
        else:
            zero_timestamp = dt.datetime.utcfromtimestamp(0)
            if impl_processor is None:
                def process(value):
                    if value is None:
                        return None
                    return zero_timestamp + value
            else:
                def process(value):
                    if value is None:
                        return None
                    return impl_processor(zero_timestamp + value)
            return process

    def result_processor(self, dialect):
        impl_processor = self.impl.result_processor(dialect)
        if self.__hasNativeImpl(dialect):
            return impl_processor
        else:
            zero_timestamp = dt.datetime.utcfromtimestamp(0)
            if impl_processor is None:
                def process(value):
                    if value is None:
                        return None
                    return value - zero_timestamp
            else:
                def process(value):
                    if value is None:
                        return None
                    return impl_processor(value) - zero_timestamp
            return process

class FLOAT(Float): pass

TEXT = Text
class NUMERIC(Numeric): pass
class DECIMAL(Numeric): pass
class INT(Integer): pass
INTEGER = INT
class SMALLINT(Smallinteger): pass
class TIMESTAMP(DateTime): pass
class DATETIME(DateTime): pass
class DATE(Date): pass
class TIME(Time): pass
class CLOB(Text): pass
class VARCHAR(String): pass
class CHAR(String): pass
class NCHAR(Unicode): pass
class BLOB(Binary): pass
class BOOLEAN(Boolean): pass

NULLTYPE = NullType()

# using VARCHAR/NCHAR so that we dont get the genericized "String"
# type which usually resolves to TEXT/CLOB
type_map = {
    str : VARCHAR,
    unicode : NCHAR,
    int : Integer,
    float : Numeric,
    dt.date : Date,
    dt.datetime : DateTime,
    dt.time : Time,
    dt.timedelta : Interval,
    type(None): NullType
}