Python sqlalchemy.types 模块，TIMESTAMP 实例源码

def _set_orderby_desc(self, query, model, limit,
                          last_id, offset, descending, orderby):
        """Return an updated query with the proper orderby and desc."""
        if orderby == 'fav_user_ids':
            n_favs = func.coalesce(func.array_length(model.fav_user_ids, 1), 0).label('n_favs')
            query = query.add_column(n_favs)
        if orderby in ['created', 'updated', 'finish_time']:
            if descending:
                query = query.order_by(desc(
                                            cast(getattr(model,
                                                         orderby),
                                                 TIMESTAMP)))
            else:
                query = query.order_by(cast(getattr(model, orderby), TIMESTAMP))
        else:
            if orderby != 'fav_user_ids':
                if descending:
                    query = query.order_by(desc(getattr(model, orderby)))
                else:
                    query = query.order_by(getattr(model, orderby))
            else:
                if descending:
                    query = query.order_by(desc("n_favs"))
                else:
                    query = query.order_by("n_favs")
        if last_id:
            query = query.limit(limit)
        else:
            query = query.limit(limit).offset(offset)
        return query

def get_col_spec(self):
        return "TIMESTAMP"

def sqltype_to_stdtype(sqltype):
    import sqlalchemy.types as sqltypes
    if isinstance(sqltype, (sqltypes.VARCHAR, sqltypes.CHAR, sqltypes.TEXT, sqltypes.Enum, sqltypes.String)):
        return _STRING_TYPE
    if isinstance(sqltype, (sqltypes.DATETIME, sqltypes.DATE, sqltypes.TIME, sqltypes.TIMESTAMP)):
        return _DATE_TYPE
    if isinstance(sqltype, (sqltypes.INTEGER, sqltypes.BIGINT, sqltypes.SMALLINT, sqltypes.Integer)):
        return _INTEGER_TYPE
    if isinstance(sqltype, (sqltypes.REAL, sqltypes.DECIMAL, sqltypes.NUMERIC, sqltypes.FLOAT)):
        return _DECIMAL_TYPE
    if isinstance(sqltype, sqltypes.BOOLEAN):
        return _BOOLEAN_TYPE

def stdtype_to_sqltype(stdtype):
    import sqlalchemy.types as sqltypes
    if isinstance(stdtype, stdtypes.StringType):
        return sqltypes.VARCHAR(length=stdtype.max_len) if 0 < stdtype.max_len < 65536 else sqltypes.TEXT()
    if isinstance(stdtype, stdtypes.BoolType):
        return sqltypes.BOOLEAN()
    if isinstance(stdtype, stdtypes.DateType):
        return sqltypes.DATE() if stdtype.only_date else sqltypes.TIMESTAMP()
    if isinstance(stdtype, stdtypes.IntegerType):
        return sqltypes.BIGINT() if stdtype.length > 11 else sqltypes.INTEGER()
    if isinstance(stdtype, stdtypes.DecimalType):
        return sqltypes.DECIMAL()
    if isinstance(stdtype, stdtypes.ArrayType):
        return sqltypes.ARRAY(item_type=stdtype.item_type)

def test_reflect_select(engine, table):
    assert len(table.c) == 9
    assert isinstance(table.c.integer, Column)
    assert isinstance(table.c.integer.type, types.Integer)
    assert isinstance(table.c.timestamp.type, types.TIMESTAMP)
    assert isinstance(table.c.string.type, types.String)
    assert isinstance(table.c.float.type, types.Float)
    assert isinstance(table.c.boolean.type, types.Boolean)
    assert isinstance(table.c.date.type, types.DATE)
    assert isinstance(table.c.datetime.type, types.DATETIME)
    assert isinstance(table.c.time.type, types.TIME)
    assert isinstance(table.c.bytes.type, types.BINARY)

    rows = table.select().execute().fetchall()
    assert len(rows) == 1000