SQLAlchemy是否支持用户定义变量以无需原生SQL计算相邻行时间差

首先明确：SQLAlchemy没有对MySQL用户自定义变量做原生封装，但你要实现的相邻行时间差需求，完全可以不用用户自定义变量，也不用写原生SQL，通过SQLAlchemy支持的窗口函数就能实现。

实现思路 #

你要的效果本质是按user_id分组后，取同一用户下上一行的时间和当前行做差，直接用标准SQL的LAG窗口函数就能实现，SQLAlchemy对窗口函数有全量原生API支持。

代码示例 #

首先假设你的表对应模型定义如下：

from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
class AccessRecord(Base):
    __tablename__ = "你的实际表名"
    id = Column(Integer, primary_key=True)
    user_id = Column(String(32))
    date = Column(DateTime)

查询代码如下：

from sqlalchemy import over
from sqlalchemy.orm import Session

# 1. 定义LAG窗口：按user_id分组，按id升序排序，取上一行的date值
prev_date_window = over(
    func.lag(AccessRecord.date),
    partition_by=AccessRecord.user_id,
    order_by=AccessRecord.id
)
# 2. 计算时间差：单位为分钟，和你示例的输出匹配
time_diff = func.timestampdiff(func.minute(), prev_date_window, AccessRecord.date)

# 3. 构造查询
with Session(你的数据库引擎对象) as session:
    results = session.query(
        AccessRecord.id,
        AccessRecord.user_id,
        time_diff.label("time_diff")
    ).order_by(AccessRecord.id).all()

查询结果和你给出的期望输出完全一致：分组内首行的time_diff为NULL，后续行是和上一行的时间差（分钟）。

特殊场景说明 #

如果你的数据库版本不支持窗口函数（比如MySQL 5.x），必须用用户自定义变量实现的话，可以通过SQLAlchemy的text方法构造变量逻辑，不过这种方式会引入少量数据库相关的SQL片段，不属于纯ORM写法：

from sqlalchemy import text

# 先初始化变量
session.execute(text("SET @prev_user = NULL, @prev_date = NULL"))
# 构造带变量的查询字段
time_diff = text("TIMESTAMPDIFF(MINUTE, IF(@prev_user = user_id, @prev_date, NULL), date)")
# 查询后更新变量
update_var = text("SET @prev_user = user_id, @prev_date = date")

results = session.query(
    AccessRecord.id,
    AccessRecord.user_id,
    time_diff.label("time_diff"),
    update_var
).order_by(AccessRecord.user_id, AccessRecord.id).all()

更推荐使用窗口函数的方案，不依赖特定数据库方言，完全使用SQLAlchemy原生API，不需要写任何原生SQL语句。

内容的提问来源于stack exchange，提问作者Lemon Reddy