在执行查询之前,在SQLAlchemy模型中将datetime转换为unix时间戳？

Question

我使用SQLAlchemy来处理使用奇怪时间戳格式的远程数据库 - 它将时间戳存储为自纪元以来的双精度毫秒.我想使用python datetime对象,所以我在我的模型中编写了getter/setter方法,遵循以下要点:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import synonym
from sqlalchemy.dialects.mysql import DOUBLE
import datetime

Base = declarative_base()
class Table(Base):
    __tablename__ = "table"

    id = Column(Integer, primary_key=True)
    _timestamp = Column("timestamp", DOUBLE(asdecimal=False))

    @property
    def timestamp(self):
        return datetime.datetime.utcfromtimestamp(float(self._timestamp)/1000.)

    @timestamp.setter
    def timestamp(self, dt):
        self._timestamp = float(dt.strftime("%s"))*1000.

    timestamp = synonym('_timestamp', descriptor=timestamp)

这非常适合在表中插入新行并使用表中的对象:

>>> table = session.query(Table).first()
<Table id=1>
>>> table.timestamp
datetime.datetime(2016, 6, 27, 16, 9, 3, 320000)
>>> table._timestamp
1467043743320.0

但是,当我尝试在过滤器表达式中使用datetime时,它会崩溃:

>>> july = datetime.datetime(2016, 7, 1)
>>> old = session.query(Table).filter(Table.timestamp < july).first()
/lib/python2.7/site-packages/sqlalchemy/engine/default.py:450: Warning: Truncated incorrect DOUBLE value: '2016-07-01 00:00:00'
>>> july_flt = float(july.strftime("%s"))*1000.
>>> old = session.query(Table).filter(Table.timestamp < july_flt).first()
<Table id=1>

我假设这是因为我的getter/setter方法适用于表类的实例,但不改变类本身的行为.我尝试使用混合属性而不是同义词重写:

from sqlalchemy.ext.hybrid import hybrid_property

class Table(Base):
    __tablename__ = "table"

    id = Column(Integer, primary_key=True)
    _timestamp = Column("timestamp", DOUBLE(asdecimal=False))

    @hybrid_property
    def timestamp(self):
        return datetime.datetime.utcfromtimestamp(float(self._timestamp)/1000.)

    @timestamp.setter
    def timestamp(self, dt):
        self._timestamp = float(dt.strftime("%s"))*1000.

同样,这适用于Table实例,但在查询上失败 - 现在它在我运行查询时遇到了我的getter方法:

>>> july = datetime.datetime(2016, 7, 1)
>>> old = session.query(Table).filter(Table.timestamp < july).first()
Traceback:
  File "models.py", line 42, in timestamp
    return datetime.datetime.utcfromtimestamp(float(self._timestamp)/1000.)
TypeError: float() argument must be a string or a number

使用调试器,我可以看到getter正在接收Table._timestamp类(不是特定的Table._timestamp,而不是'7月').

我看到我可以使用hybrid_property.expression装饰器来定义用于将时间戳转换为datetime的SQL表达式,但我真正喜欢的是将日期时间转换为python端的时间戳,然后使用时间戳运行查询.换句话说,我想在任何地方使用日期时间(包括在查询中),但是在SQL端使用微秒时间戳完成所有操作.我怎样才能做到这一点？

Answer 1

您必须使用自定义类型，这并不像听起来那么可怕。

from sqlalchemy.types import TypeDecorator


class DoubleTimestamp(TypeDecorator):
    impl = DOUBLE

    def __init__(self):
        TypeDecorator.__init__(self, as_decimal=False)

    def process_bind_param(self, value, dialect):
        return value.replace(tzinfo=datetime.timezone.utc).timestamp() * 1000

    def process_result_value(self, value, dialect):
        return datetime.datetime.utcfromtimestamp(value / 1000)

然后Table变成：

class Table(Base):
    __tablename__ = "table"

    id = Column(Integer, primary_key=True)
    timestamp = Column(DoubleTimestamp)

然后你提到的一切都有效。您插入、选择并与datetimes进行比较，但它存储为DOUBLE.

在这里，我使用了不同的逻辑在时间戳之间进行转换，因为strftime('%s')这不是正确的解决方案。这是一个不同的问题，已在此处正确回答。哦，我注意到你说微秒，但只在你发布的代码中转换为毫秒，除非是口误。