SQLAlchemy DateTime时区过滤异常排查及优化需求

问题描述 #

尝试使用SQLAlchemy依据时区过滤数据，但当前查询仅基于datetime对象的值筛选，未考虑时区信息。插入带时区的DateTime记录后，预期查询能精准匹配指定时区的datetime值，但实际仅匹配datetime的数值部分，无法按时区筛选。

原代码 #

from sqlalchemy import Column, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import sessionmaker
import datetime
import pytz

engine = create_engine('sqlite:///mydatabase121.db', echo=True)
Base = declarative_base()

# 创建表结构
class MyTable(Base):
    __tablename__ = 'my_table'
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime(timezone=True))


Base.metadata.create_all(engine)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 插入带时区的测试数据
# US/Eastern时区2024-01-01 10:00
created_at = pytz.timezone('US/Eastern').localize(datetime.datetime(2024, 1, 1, 10, 0))
record = MyTable(created_at=created_at)
session.add(record)
session.commit()
# UTC时区2024-01-01 11:00
created_at = pytz.timezone('UTC').localize(datetime.datetime(2024, 1, 1, 11, 0))
record = MyTable(created_at=created_at)
session.add(record)
session.commit()
# US/Eastern时区2024-01-01 12:00
created_at = pytz.timezone('US/Eastern').localize(datetime.datetime(2024, 1, 1, 12, 0))
record = MyTable(created_at=created_at)
session.add(record)
session.commit()


# 尝试按指定时区筛选数据
records = session.query(MyTable).filter(MyTable.created_at == pytz.timezone('US/Eastern').localize(datetime.datetime(2024, 1, 1, 10, 0))).all()

# 打印查询结果
print("\nPrinting Record : \n")
for record in records:
    print(record.id, record.created_at)


# 时间时区转换函数
def convert_datetime_timezone(dt, tz1, tz2):
    tz1 = pytz.timezone(tz1)
    tz2 = pytz.timezone(tz2)
    dt = dt.astimezone(tz1)
    dt = tz2.normalize(dt.astimezone(tz2))
    dt = dt.strftime("%Y-%m-%d %H:%M:%S")

    return dt

converted = convert_datetime_timezone(records[0].created_at, 'US/Eastern', 'UTC')
converted2 = convert_datetime_timezone(records[0].created_at, 'US/Eastern', 'Asia/Kolkata')

# 打印转换结果
print("\nDatetime Converted in UTC Timezone: \n")
print(converted)
print("\nDatetime Converted in Indian Timezone: \n")
print(converted2)

补充测试代码：

records = session.query(MyTable).filter(
       MyTable.created_at == pytz.timezone(
       'US/Eastern').localize(datetime.datetime(2025, 1, 1, 11, 0))
   ).all()

# 打印查询结果
print("\nPrinting Record : \n")
for record in records:
    print(record.id, record.created_at)

问题分析 #

SQLite对时区原生支持有限，当使用DateTime(timezone=True)时，SQLAlchemy会自动将带时区的datetime转换为UTC时间存储。直接用带时区的datetime对象等值匹配时，实际是比较存储的UTC时间与目标时间转换后的UTC值，但如果需求是筛选指定时区下的某个时间点记录，而非UTC时间匹配，就需要调整查询逻辑。

解决方案 #

方法一：转换目标时间为UTC后查询 #

利用数据库存储UTC时间的特性，将目标时区的时间转换为UTC后再匹配，高效且直接：

# 定义目标时区和时间
target_tz = pytz.timezone('US/Eastern')
target_local_time = target_tz.localize(datetime.datetime(2024, 1, 1, 10, 0))
# 转换为UTC时间
target_utc_time = target_local_time.astimezone(pytz.utc)
# 执行查询
records = session.query(MyTable).filter(MyTable.created_at == target_utc_time).all()

方法二：数据库层面转换时区后匹配 #

使用SQLAlchemy的func.timezone函数，将数据库中的UTC时间转换为目标时区后再与本地时间匹配，适合范围查询（如筛选目标时区某天的所有记录）：

from sqlalchemy import func

# 筛选US/Eastern时区下2024-01-01 10:00的记录
records = session.query(MyTable).filter(
    func.timezone('US/Eastern', MyTable.created_at) == datetime.datetime(2024, 1, 1, 10, 0)
).all()

优化时间转换函数 #

从数据库取出的created_at已带时区信息，无需手动指定原时区，简化转换逻辑：

def convert_datetime_timezone(dt, target_tz_name):
    target_tz = pytz.timezone(target_tz_name)
    converted_dt = dt.astimezone(target_tz)
    return converted_dt.strftime("%Y-%m-%d %H:%M:%S")

# 使用示例
converted = convert_datetime_timezone(records[0].created_at, 'UTC')
converted2 = convert_datetime_timezone(records[0].created_at, 'Asia/Kolkata')

内容的提问来源于stack exchange，提问作者Pankaj Chowdhury