如何用Python+SQLAlchemy一次查询PostgreSQL时序分区销售数据

嘿，这个场景我太熟了！完全不用跑6次查询，咱们用SQLAlchemy的分组查询结合case表达式，一次就能把所有6个区间的销售数据捞出来，效率还高得多。下面给你两种方案，按需选择：

方案1：按星期几映射区间（灵活适配每周） #

如果你的区间是固定的「星期X-星期Y」模式，比如周六-周日、周日-周一这类，推荐用PostgreSQL的时间函数把Unix时间戳转成星期几，再用case映射到对应的区间标签，最后分组统计。

步骤1：导入必要的SQLAlchemy工具 #

from sqlalchemy import func, case, and_

步骤2：构建区间标签映射 #

先把Unix时间戳转成PostgreSQL的时间类型，再提取星期几（注意PostgreSQL里dow字段：0=周日，1=周一，...，6=周六），然后用case把每个星期几对应到你的目标区间：

# 定义每个星期几对应的区间标签
interval_label = case(
    [(func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 6, '周六-周日'),
     (func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 0, '周日-周一'),
     (func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 1, '周一-周二'),
     (func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 2, '周二-周三'),
     (func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 3, '周三-周四'),
     (func.extract('dow', func.to_timestamp(SalesTable.when_in_unix_time)) == 4, '周四-周五')],
    else_='周五-周六'  # 可以根据你的需求调整，或者设为None忽略
)

步骤3：执行分组查询 #

指定整个统计周期的起止时间，然后按区间标签分组，统计销售数据（这里以统计销售笔数为例，你也可以换成func.sum(SalesTable.amount)统计金额）：

# 先定义整个一周的时间范围（比如最近7天的起止Unix时间戳）
overall_start_time = ...  # 你需要的一周起始Unix时间
overall_end_time = ...    # 你需要的一周结束Unix时间

# 构建查询
query = db.query(
    interval_label.label('interval'),
    func.count(SalesTable.id).label('sales_count'),
    # 如果需要每个区间的实际起止时间，可以加下面两行
    func.min(SalesTable.when_in_unix_time).label('interval_start'),
    func.max(SalesTable.when_in_unix_time).label('interval_end')
).filter(
    SalesTable.when_in_unix_time >= overall_start_time,
    SalesTable.when_in_unix_time <= overall_end_time
).group_by('interval').order_by('interval')

# 获取结果
results = query.all()

返回的results里每个元素是一个元组，包含区间名称、销售数量、区间起止时间（如果加了的话），直接就能用来生成图表了。

方案2：按固定时间范围匹配区间（精准指定每个区间的时间） #

如果你的区间是特定的时间范围（比如某一周的周六0点到周日24点这种固定值），可以直接用case匹配每个时间区间：

# 提前计算好每个区间的起止Unix时间戳
sat_sun_start = ...
sat_sun_end = ...
sun_mon_start = ...
sun_mon_end = ...
# 其余4个区间同理...

interval_label = case(
    [(and_(SalesTable.when_in_unix_time >= sat_sun_start, SalesTable.when_in_unix_time <= sat_sun_end), '周六-周日'),
     (and_(SalesTable.when_in_unix_time >= sun_mon_start, SalesTable.when_in_unix_time <= sun_mon_end), '周日-周一'),
     # 其余4个区间的条件...
    ]
)

# 后面的分组查询逻辑和方案1一致
query = db.query(
    interval_label.label('interval'),
    func.count(SalesTable.id).label('sales_count')
).group_by('interval').order_by('interval')

results = query.all()

补充：处理空区间（保证所有6个区间都有结果） #

如果某个区间没有销售数据，上面的查询不会返回该区间的行。如果你的图表需要显示所有6个区间（哪怕销售数为0），可以用左连接的方式，先生成所有区间的列表，再关联统计结果：

from sqlalchemy.sql import text

# 生成包含所有6个区间的临时数据集
all_intervals = db.query(text("""
    SELECT '周六-周日' AS interval UNION ALL
    SELECT '周日-周一' AS interval UNION ALL
    SELECT '周一-周二' AS interval UNION ALL
    SELECT '周二-周三' AS interval UNION ALL
    SELECT '周三-周四' AS interval UNION ALL
    SELECT '周四-周五' AS interval
""")).subquery()

# 先获取统计结果的子查询
stats_subquery = db.query(
    interval_label.label('interval'),
    func.count(SalesTable.id).label('sales_count')
).filter(
    SalesTable.when_in_unix_time >= overall_start_time,
    SalesTable.when_in_unix_time <= overall_end_time
).group_by('interval').subquery()

# 左连接，保证所有区间都被返回，空区间用0填充
final_query = db.query(
    all_intervals.c.interval,
    func.coalesce(stats_subquery.c.sales_count, 0).label('sales_count')
).outerjoin(stats_subquery, all_intervals.c.interval == stats_subquery.c.interval).order_by(all_intervals.c.interval)

full_results = final_query.all()

这样返回的full_results里，每个区间都会有对应的销售数，没有销售的区间会显示0，完美适配绘图需求。

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