如何在有和没有 SQLAlchemy 的情况下处理 MySQL 上的时区感知日期时间对象的“超出范围”警告

Question

在 MySQL 上的 DateTime 列中插入可识别时区的日期时间对象时，我收到来自 Mysql-Python 的警告：

test_mysql.py:13: Warning: Out of range value for column 'created_at' at row 1
  cur.execute("INSERT INTO test (created_at) VALUES (%s)", now)

测试代码如下：

import MySQLdb
from datetime import datetime
from pytz import utc

conn = MySQLdb.connect(...)  # connect
cur = conn.cursor()
now = datetime.utcnow()
cur.execute("CREATE TABLE test (created_at DATETIME)")
print("Test 1")
cur.execute("INSERT INTO test (created_at) VALUES (%s)", now)
now = utc.localize(now)
print("Test 2")
cur.execute("INSERT INTO test (created_at) VALUES (%s)", now)
print("Test 3")
now = now.replace(tzinfo=None)
assert now.tzinfo is None
cur.execute("INSERT INTO test (created_at) VALUES (%s)", now)
print("Tests done")
cur.execute("DROP TABLE test")

完整输出为：

Test 1
Test 2
test_mysql.py:13: Warning: Out of range value for column 'created_at' at row 1
  cur.execute("INSERT INTO test (created_at) VALUES (%s)", now)
Test 3
Tests done

我在我的原始应用程序中使用 SQLAlchemy，但由于这是 SQLAlchemy 下的问题，我对使用 SA 和不使用 SA 的解决方案感兴趣。

Answer 1

如果没有 SQLAlchemy，答案基本上已经在问题中了：让它变得天真，但以这样一种方式，你总是存储相同的（阅读：UTC）时区：

if now.tzinfo is not None:
    now = now.astimezone(utc).replace(tzinfo=None)
return now

这将确保始终存储 UTC（但如果传递 naive，请确保它总是 naive！并且当将项目从数据库中拉出时，请确保它能够识别 UTC再次：

if created_at.tzinfo is None:
    created_at = utc.localize(created_at)
return created_at

由于 MySQL 无法识别 tz 对象，因此可以删除检查（应始终为 None）。

对于 SQLAlchemy，可以使用相同的方法（但扩展了让 SA 自动执行此操作）。为此，我们使用TypeDecorator：

class TZDateTime(TypeDecorator):
    """
    Coerces a tz-aware datetime object into a naive utc datetime object to be
    stored in the database. If already naive, will keep it.

    On return of the data will restore it as an aware object by assuming it
    is UTC.

    Use this instead of the standard :class:`sqlalchemy.types.DateTime`.
    """

    impl = DateTime

    def process_bind_param(self, value, dialect):
        if value.tzinfo is not None:
            value = value.astimezone(utc).replace(tzinfo=None)
        return value

    def process_result_value(self, value, dialect):
        if value.tzinfo is None:
            value = utc.localize(value)
        return value

这样可以确保处理的数据在 UTC 上始终是 tz 感知的，只要始终为 UTC 生成幼稚值：例如 datetime.datetime.utcnow()。

定义列时，使用它而不是常规类型：

class Test(Base):
    __tablename__ = 'test'
    created_at = Column(TZDateTime)