python - 使用带有 mysql unix 时间戳的 sqlalchemy 定义表

Question

背景，有几种方法可以在 MySQ 中存储日期。

1. 作为字符串，例如“09/09/2009”。
2. 作为使用函数 UNIX_TIMESTAMP() 的整数，这应该是传统的 unix 时间表示（您知道自纪元以来的秒数加/减闰秒）。
3. 作为 MySQL TIMESTAMP，mysql 特定的数据类型与 unix 时间戳不同。

4. 作为 MySQL 日期字段，另一种 mysql 特定数据类型。

   不要将案例 2 与案例 3（或案例 4）混淆，这一点非常重要。我有一个带有整数日期字段的现有表（案例 2）我如何在 sqlalchemy 中以我不必访问 mysql 的“FROM_UNIXTIME”函数的方式定义它？

   作为记录，仅使用 sqlalchemy.types.DateTime 并希望它在检测到整数列不起作用时做正确的事情，它适用于时间戳字段和日期字段。

Answer 1

我认为您展示的类型装饰器存在一些问题。

1. impl应该sqlalchemy.types.Integer代替DateTime.
2. 装饰器应该允许可以为空的列。

这是我的想法：

import datetime, time
from sqlalchemy.types import TypeDecorator, DateTime, Integer

class IntegerDateTime(TypeDecorator):
    """a type that decorates DateTime, converts to unix time on
    the way in and to datetime.datetime objects on the way out."""
    impl = Integer # In schema, you want these datetimes to
                   # be stored as integers.
    def process_bind_param(self, value, _):
        """Assumes a datetime.datetime"""
        if value is None:
            return None # support nullability
        elif isinstance(value, datetime.datetime):
            return int(time.mktime(value.timetuple()))
        raise ValueError("Can operate only on datetime values. "
                         "Offending value type: {0}".format(type(value).__name__))
    def process_result_value(self, value, _):
        if value is not None: # support nullability
            return datetime.datetime.fromtimestamp(float(value))

Answer 2

所以，是的，这种方法有效。最后我回答了我自己的问题：/，希望有人觉得这很有用。

import datetime, time
from sqlalchemy.types import TypeDecorator, DateTime
class IntegerDateTime(TypeDecorator):
    """a type that decorates DateTime, converts to unix time on
    the way in and to datetime.datetime objects on the way out."""
    impl = DateTime
    def process_bind_param(self, value, engine):
        """Assumes a datetime.datetime"""
        assert isinstance(value, datetime.datetime)
        return int(time.mktime(value.timetuple()))
    def process_result_value(self, value, engine):
        return datetime.datetime.fromtimestamp(float(value))
    def copy(self):
        return IntegerDateTime(timezone=self.timezone)