SQLalchemy:自动对列应用过滤器

Question

我想使用自定义SQLalchemy数据类型存储大量的python浮点时间戳作为标准postgresql时间戳.我看到了可能的解决方案,例如使用带有mysql unix时间戳的sqlalchemy定义表.我不喜欢它,它是相当低效的,因为它将浮点数转换为时间对象,然后返回到时间戳,而这应该只是"乘以1e6,添加常量并转换为int" - 转换,应该在时间上接近一个noop.而

%timeit float(calendar.timegm(datetime.datetime.utcfromtimestamp(1542098001).timetuple()))

在我的机器上每次通话浪费了2us.使用postgresql to_timestamp,extract(epoch from timestamp_col)应该更快,更适合分析工作负载.SQLalchemy中是否有一种方法可以自动对SQL级别的那些转换应用于影响该列的所有语句？我还考虑使用一个简单的浮点字段进行存储,但我更喜欢能够使用时间函数来访问数据.

import time
from datetime import datetime
from calendar import timegm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, String, Integer, DateTime, TypeDecorator, create_engine


class Timestamp(TypeDecorator):
    impl = DateTime

    def process_bind_param(self, value, dialect):
        if value is None:
            return
        elif isinstance(value, datetime):
            return value
        elif isinstance(value, (int, float)):
            return datetime.utcfromtimestamp(value)
        raise TypeError("Any form of time object required, but received {}".format(type(value).__name__))

    def process_result_value(self, value, dialect):
        if value is None:
            return None # support nullability
        elif isinstance(value, datetime):
            return float(timegm(value.timetuple()))
        raise vn.ValidationTypeError("datetime object required, but received {}".format(type(value).__name__))


Base = declarative_base()

class SystemLog(Base):
    __tablename__ = 'systemlog'
    id = Column(Integer, primary_key=True)
    time = Column(Timestamp, index=True, nullable=False, default=datetime.utcnow)
    type = Column(String(20))
    message = Column(String(2000))


engine = create_engine('sqlite://')
Session = sessionmaker(bind=engine)

s = Session()
Base.metadata.create_all(engine)
s.commit()

start_time = time.time()
sample_data = [SystemLog(time=start_time + i) for i in xrange(10000)]
s.add_all(sample_data)
s.commit()