具有特定格式的 SQLAlchemy ORM DateTime

Question

我有一个表，它以以下格式存储日期时间值：（例如）20160213145512。不幸的是，SQLAlchemy 没有机会指定用于将日期时间值存储到具有 sqlalchemy 的 DateTime 类型的数据库中的格式。

因此，我创建了以下数据类型：

class Timestamp(TypeDecorator):
    impl = String(14)

    def process_bind_param(self, value, dialect):
        if isinstance(value, datetime):
            value = '%04d%02d%02d%02d%02d%02d' % (value.year,
                                                  value.month,
                                                  value.day,
                                                  value.hour,
                                                  value.minute,
                                                  value.second)
        return value


    def process_result_value(self, value, dialect):
        if value:
            return datetime(year=int(value[0:4]),
                            month=int(value[4:6]),
                            day=int(value[6:8]),
                            hour=int(value[8:10]),
                            minute=int(value[10:12]),
                            second=int(value[12:14])
                            )

出于性能原因，我不使用 strftime 和 strptime。 我现在能做的是：

class TestTable(Base):
    __tablename__ = "TESTTABLE"

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

通过日期时间对象定义日期时间来插入行：

session.add(TestTable(timestamp=datetime(year=2016, month=2, day=3, hour=2)))

或者通过将日期时间定义为字符串来插入一行：

session.add(TestTable(timestamp="20160203031422"))

到目前为止，一切都很好。 但以下内容未按预期工作：

session.query(TestTable).filter(TestTable.timestamp.year == 2016).all()

如果我正在运行此代码 sn-p 我会收到此错误：

AttributeError: Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with ORDERS.dz_statakt has an attribute 'year'

但这是可行的：

session.query(TestTable).filter(TestTable.timestamp.startswith("2016")).all()

直到现在我都无法解决这个问题。我想我在自己的 Timestamp 类中遗漏了一些东西。

这可能吗？

Answer 1

这是可能的，但有点复杂。

首先，您的代码不起作用的原因。 TestTable.timestamp 是 Column。任何地方都没有定义year 属性。 year 属性位于从 process_result_value 返回的 datetime 对象上。 TestTable.timestamp.startswith 起作用的原因是因为.startswith 是considered an operator by SQLAlchemy。

因此，要使其正常工作，您需要在 TestTable.timestamp 上定义属性，以便在将它们与值进行比较时为您提供正确的 SQL 表达式：

class Timestamp(TypeDecorator):
    impl = String(14)

    class comparator_factory(TypeDecorator.Comparator, impl.comparator_factory):
        @property
        def year(self):
            return cast(func.substr(self.expr, 1, 4), Integer)

        ...

    def process_bind_param(self, value, dialect):
        ...

    def process_result_value(self, value, dialect):
        ...