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

Question

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

作为字符串的"09/09/2009".

• As

• ，例如使用UNIX_TIMESTAMP()函数的整数--这应该是传统的UNIX_TIMESTAMP时间表示形式(您知道自时代加/减闰秒以来的秒数)。

• 作为MySQL时间戳，它的特定数据类型与unix时间戳不一样。

• 作为mysql日期字段，另一种特定的数据类型。

非常重要的是不要混淆案例2和案例3(或案例4)。我有一个带有整数日期字段(例2)的现有表，如何在sqlalchemy中定义它，而不必访问mysql的"FROM_UNIXTIME“函数？

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

Answer 1

我想你展示的装潢师有几个问题。

1. impl应该是sqlalchemy.types.Integer而不是DateTime.
2. The装饰符，应该允许空列。

以下是我的想法：

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)