在执行查询之前将datetime转换为SQLAlchemy模型中的unix timestamp？

Question

我正在使用SQLAlchemy与使用奇怪时间戳格式的远程数据库一起工作 - 它将时间戳存储为自时代以来的双精度毫秒。我想使用Python datetime对象的工作，所以我写在我的模型getter/setter方法，下面this gist：

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 

然而，它打破了，当我尝试在筛选表达式中使用日期时间：

>>> 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方法适用ŧ o表类的实例，但不要改变类本身的行为。我试着用的，而不是一个代名词混合属性改写：

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. 

再次，这与工作表的情况下，但未能对查询 - 现在它打我的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装饰器来定义一个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) 

然后一切你所提到的作品。您插入，选择并与datetime s比较，但它存储为DOUBLE。

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