SQLAlchemy聚合查询报错：如何实现指定时间范围的最值、平均值分页？

你的报错根源是my_table.select()的用法错误——直接把聚合函数传给这个方法会导致SQLAlchemy错误地将聚合逻辑放到WHERE子句中，违反SQL语法规则。以下是基于现有Table模式的修正方案：

修正后的代码 #

from sqlalchemy import func, select  # 新增select导入
...

engine = create_engine(URL)
metadata = MetaData()

my_table = Table(
    "my_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("value", Numeric()),
    Column("timestamp", DateTime(timezone=True)),
)

database = Database(DATABASE_URL)
metadata.create_all(engine)

# 正确构造查询：用select()指定聚合字段，添加时间筛选、分页
query = (
    select(
        func.min(my_table.c.value).label("min_value"),
        func.max(my_table.c.value).label("max_value"),
        func.avg(my_table.c.value).label("avg_value")
    )
    .where(my_table.c.timestamp > some_date)
    .where(my_table.c.timestamp < some_other_date)
    .limit(100)  # 单页数据条数
    .offset(0)   # 分页偏移量，按需调整
)
result = database.fetch_all(query=query)

关键修正说明 #

• 改用select()构造器：这是SQLAlchemy Core中构造聚合查询的标准方式，能确保聚合函数被正确放到SELECT子句中。
• 给聚合结果加label()：方便后续通过别名直接获取对应统计值，比如result[0].min_value。
• 完全保留原有Table模式：没有使用Session、ORM或原生SQL，完全基于你提供的表定义实现需求。

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