SqlAlchemy使用from_statement时自定义类型process_result_value返回值不生效问题

问题原因 #

这一现象不属于预期的正常行为，是SQLAlchemy 1.3版本处理原生SQL查询与ORM自定义类型映射的已知特性：当使用from_statement配合无类型标注的text()原生SQL时，ORM层会优先读取数据库驱动返回的原生值类型，即使process_result_value方法被触发执行，最终实体属性赋值环节仍可能出现类型覆盖问题，导致时区信息丢失。

解决方案 #

方案1：给原生SQL的时间列显式绑定自定义类型 #

在text()语句后通过columns()方法指定时间列使用你实现的UTCDateTime类型，强制SQLAlchemy使用自定义类型的处理结果：

from sqlalchemy import text

# 假设你需要映射的时间字段为create_time
stmt = text("SELECT * from table where ...")
stmt = stmt.columns(create_time=UTCDateTime)
result = db.query(User).from_statement(stmt).first()

该方案无需修改现有逻辑，适配性最强。

方案2：升级SQLAlchemy版本 #

这个类型映射的问题在SQLAlchemy 1.4及以上版本已经被官方修复，升级后原生SQL查询的自定义类型处理逻辑会和普通ORM查询保持一致，无需额外配置。

方案3：添加ORM加载事件钩子补全时区 #

如果不能升级版本也不想修改每个查询语句，可以通过ORM的load事件给所有时间字段自动补全时区信息：

from sqlalchemy import event
from datetime import timezone

@event.listens_for(User, 'load')
def fill_utc_tz(target, _):
    # 遍历所有时间类型字段补全时区
    for col in target.__table__.columns:
        if isinstance(col.type, UTCDateTime):
            val = getattr(target, col.key)
            if val and val.tzinfo is None:
                setattr(target, col.key, val.replace(tzinfo=timezone.utc))

内容的提问来源于stack exchange，提问作者Elias Dorneles