如何用SQLAlchemy ORM单查询获取多范围下某列的最大值

1. 定义ORM模型 #

假设你的数据表名为data_records，包含日期时间列datetime_col和数值列value_col，ORM模型定义如下：

from sqlalchemy import Column, DateTime, Float, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class DataRecord(Base):
    __tablename__ = "data_records"
    id = Column(Integer, primary_key=True)
    datetime_col = Column(DateTime, nullable=False)
    value_col = Column(Float, nullable=False)

2. 构造预定义时间区间的子查询 #

通过union_all将多个时间区间合并为一个子查询，这是解决问题的核心部分：

from sqlalchemy import select, union_all, func, literal_column
from sqlalchemy.orm import Session

# 替换成你的实际预定义时间区间
intervals = [
    ("2024-01-01 00:00:00", "2024-01-01 12:00:00"),
    ("2024-01-01 12:00:00", "2024-01-02 00:00:00"),
    ("2024-01-02 00:00:00", "2024-01-02 12:00:00"),
]

# 为每个区间生成单独的select语句
interval_selects = []
for start, end in intervals:
    # 如果数据库日期格式需要转换，可使用func.str_to_date处理
    interval_select = select(
        literal_column(f"'{start}'").label("start_time"),
        literal_column(f"'{end}'").label("end_time")
    )
    interval_selects.append(interval_select)

# 用union_all合并所有区间查询，生成子查询
intervals_subq = union_all(*interval_selects).subquery("intervals")

3. 关联主表查询区间最大值 #

将区间子查询与数据主表关联，过滤时间范围后分组计算最大值：

# 构造主查询
query = select(
    intervals_subq.c.start_time,
    intervals_subq.c.end_time,
    func.max(DataRecord.value_col).label("max_value")
).join(
    DataRecord,
    DataRecord.datetime_col.between(intervals_subq.c.start_time, intervals_subq.c.end_time)
).group_by(
    intervals_subq.c.start_time,
    intervals_subq.c.end_time
)

# 执行查询（假设已创建Session实例session）
results = session.execute(query).all()

# 处理结果
for row in results:
    print(f"[{row.start_time} - {row.end_time}] 最大值: {row.max_value}")

动态生成连续12小时区间（可选） #

如果不需要手动指定区间，而是要从最早数据开始自动生成连续12小时区间，可替换区间生成逻辑：

# 获取数据中最早的日期时间
min_datetime_subq = select(func.min(DataRecord.datetime_col)).scalar_subquery()

# 生成指定数量的连续12小时区间
num_intervals = 30
interval_selects = []
for i in range(num_intervals):
    start = func.date_add(min_datetime_subq, func.interval(i * 12, "hour"))
    end = func.date_add(min_datetime_subq, func.interval((i + 1) * 12, "hour"))
    interval_select = select(start.label("start_time"), end.label("end_time"))
    interval_selects.append(interval_select)

intervals_subq = union_all(*interval_selects).subquery("intervals")

内容的提问来源于stack exchange，提问作者Jes Wanson