SQLAlchemy中带有查询结果的联合汇总统计？答案

【问题标题】：Union summary statistics with query result in SQLAlchemy?SQLAlchemy中带有查询结果的联合汇总统计？
【发布时间】：2017-04-05 13:07:59
【问题描述】：

我有一个存储功率计读数的 PostgreSQL 表。我使用 SQLAlchemy 和 psycopg2 来查询数据库。一些大型站点可以有多个功率计，我有一个查询返回时间戳数据，按设施汇总：

原始表：

timestamp | meter_id | facility_id | reading
  1:00:00 |        1 |           1 |     1.0
  1:00:00 |        2 |           1 |     1.5
  1:00:00 |        3 |           2 |     2.1
  1:00:30 |        1 |           1 |     1.1
  1:00:30 |        2 |           1 |     1.6
  1:00:30 |        3 |           2 |     2.2

聚合：

timestamp | facility_1 | facility_2
  1:00:00 |        2.5 |       2.1
  1:00:30 |        2.7 |       2.2

我用于此的查询如下所示：

SELECT
    reading.timestamp,
    sum(reading.reading) FILTER (WHERE reading.facility_id = 1) as facility_1,
    sum(reading.reading) FILTER (WHERE reading.facility_id = 2) as facility_2
FROM reading
GROUP BY reading.timestamp
WHERE
    reading.timestamp >= 1:00:00 AND reading.timestamp < 1:01:00
    AND reading.facility_id IN 1, 2

（对于任何 SQL 错误，我很抱歉，为了清楚起见，我已经稍微简化了问题）。我经常需要对数据进行下采样以进行显示，我通过将上述查询包装在 FROM...AS... 子句中并将数据分箱到更大的时间间隔中来实现。不过，在此之前，我想从派生的“设施”表中获取一些汇总统计数据——最小读数、最大读数、平均读数等，类似于 this blog post 中描述的内容。但是，我不知道如何使用 SQLALchemy 来获取这些数据——我不断从生成的 SQL 中得到 psycopg2 错误。上述查询的我的 SQLAlchemy 版本是：

selects = [Reading.timestamp,
    sqlalchemy.func.sum(Reading.reading).filter(Reading.facility_id==1),
    sqlalchemy.func.sum(Reading.reading).filter(Reading.facility_id==2)
]
base_query = db.session.query(*selects). \
    group_by(Reading.timestamp). \
    filter(Reading.facility_id.in_([1, 2])). \
    filter(and_(Reading.timestamp>=start_time, Reading.timestamp<=end_time)). \
    order_by(Reading.timestamp)

我可以通过以下方式获得汇总统计信息：

subq = base_query.subquery()
avg_selects = [sqlalchemy.func.avg(col) for col in subq.columns]
avg_query = db.session.query(*avg_selects)

这将返回一行，其中包含原始查询中所有列的平均值。但是，我不知道如何用我的原始查询来获得这个——我最终不得不单独获取统计信息，这感觉像是一种巨大的浪费（这些查询可以跨越很多行）。像下面这样的查询总是返回错误：

all = base_query.union(avg_query).all()

ProgrammingError: (psycopg2.ProgrammingError) syntax error at or near "UNION"
LINE 4: ...reading.timestamp ORDER BY reading.timestamp UNION SELE...

我觉得我对 SQLAlchemy 的子查询系统的理解很薄弱，但是我一直无法从 SQLAlchemy 文档中的子查询教程中取得进展。想法？

【问题讨论】：

标签： postgresql sqlalchemy flask-sqlalchemy psycopg2

【解决方案1】：

错误消息中的答案是正确的——我需要将 ORDER BY 子句从子查询中删除到联合操作之外，并将其移到联合之外。我使用虚拟时间戳作为汇总统计信息，以确保它们在按时间戳排序后以可预测的顺序位于查询结果的顶部。以下代码有效：

from sqlalchemy.sql import expression, func
from datetime import datetime
from models import Reading

selects = [Reading.timestamp.label("timestamp_"),
    func.sum(Reading.reading).filter(Reading.facility_id==1),
    func.sum(Reading.reading).filter(Reading.facility_id==2)
]

base_query = db.session.query(*selects). \
    group_by(Reading.timestamp). \
    filter(Reading.facility_id.in_([1, 2])). \
    filter(and_(Reading.timestamp>=start_time, Reading.timestamp<=end_time))

subq = base_query.subquery()

avg_selects = [expression.bindparam('dummy_date', datetime(1980, 1, 1)).label("timestamp_")
avg_selects += [func.avg(col) for col in subq.columns[1:]
avg_query = db.session.query(*avg_selects)

full_query = base_query.union(avg_query).order_by(asc("timestamp_"))

我很高兴听到更优雅的方式来实现这一点。查询包装在一个函数中，该函数采用任意设施 ID 列表； “列”技巧是我想出让它与任意列一起工作的唯一方法（只要第一列始终是时间戳）。

【讨论】：