将汇总统计信息与 SQLAlchemy 中的查询结果联合起来？

Question

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

原始 table:

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... 子句中并将数据分箱到更大的时间间隔来实现。不过，在此之前，我想从我的派生 "facilities" table 中获取一些摘要统计信息——最小读数、最大读数、平均读数等，类似于 [=16] 中描述的内容=].但是，我无法弄清楚如何使用 SQLALchemy 获取此数据——我不断从生成的 SQL 中收到 psycopg2 错误。我的 SQL 上述查询的 Alchemy 版本是：

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)

这将 return 包含我原始查询中所有列的平均值的单行。但是，我无法弄清楚如何使用我的原始查询来获取它——我最终不得不单独获取统计信息，这感觉是一种巨大的浪费（这些查询可以遍历很多行）。像下面这样的查询总是 return 错误：

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 的文档中的子查询教程来看，我一直没有取得进展。想法？

Answer 1

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

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 列表； "columns" 技巧是我想出的使其适用于任意列的唯一方法（只要第一列始终是时间戳）。

将汇总统计信息与 SQLAlchemy 中的查询结果联合起来？

Union summary statistics with query result in SQLAlchemy?

postgresql

sqlalchemy

psycopg2

flask-sqlalchemy