如何用SQLAlchemy获取24小时逐小时记录统计（含空小时补0）

要实现无需Python循环的全时段统计，核心思路是在SQL层面生成完整的0-23小时序列，再与实际统计结果做左连接，用coalesce填充空值为0。以下是具体实现方案：

核心步骤 #

1. 生成包含0-23所有小时的数据集
2. 获取所有需要统计的唯一name值（若只统计单个name可跳过此步）
3. 交叉连接小时序列与name列表，得到所有可能的name+小时组合
4. 左连接实际统计结果，空值补0

完整代码示例（ORM版） #

假设你已经定义好Shapes ORM模型，数据库以PostgreSQL为例（兼容MySQL等主流数据库）：

from sqlalchemy import create_engine, select, union_all, literal_column, func, Integer, literal
from sqlalchemy.orm import sessionmaker
from your_models import Shapes  # 替换为你的模型导入路径

# 初始化数据库连接
engine = create_engine("postgresql://user:password@localhost/dbname")  # 替换为你的数据库URL
Session = sessionmaker(bind=engine)
session = Session()

# 1. 生成0-23小时的子查询（通用兼容版，无需数据库特定函数）
hours = union_all(*[select(literal_column(str(h)).label("hour")) for h in range(24)])
hours_subq = hours.subquery("hours")

# 2. 获取所有唯一的shape名称（若只统计单个name，可替换为literal('circle').label('name')）
distinct_names = select(Shapes.name.distinct()).subquery("distinct_names")

# 3. 生成所有name+小时的组合（交叉连接确保全覆盖）
name_hour_pairs = select(
    distinct_names.c.name,
    hours_subq.c.hour.cast(Integer)  # 强制转为整数，避免类型不匹配
).select_from(distinct_names.cross_join(hours_subq)).subquery("name_hour")

# 4. 统计目标时间段内的实际记录数
stats_subq = select(
    Shapes.name,
    func.extract("hour", Shapes.timestamp).cast(Integer).label("hour"),
    func.count(Shapes.id).label("total_records")
).where(
    # 筛选特定日期，比between更精准
    func.date(Shapes.timestamp) == "2022-08-15"
).group_by(Shapes.name, func.extract("hour", Shapes.timestamp)).subquery("stats")

# 5. 左连接得到全时段统计，空值填充为0
final_query = select(
    name_hour_pairs.c.name,
    name_hour_pairs.c.hour,
    func.coalesce(stats_subq.c.total_records, 0).label("total_records")
).select_from(
    name_hour_pairs.outerjoin(
        stats_subq,
        (name_hour_pairs.c.name == stats_subq.c.name) &
        (name_hour_pairs.c.hour == stats_subq.c.hour)
    )
).order_by(name_hour_pairs.c.name, name_hour_pairs.c.hour)

# 执行查询并获取结果
results = session.execute(final_query).fetchall()

# 输出示例：
# [('circle', 0, 0), ('circle', 1, 0), ..., ('circle', 15, 10), ('square', 0, 5), ...]

关键细节说明 #

• 小时序列生成：用union_all手动构造0-23小时是通用方案，若用PostgreSQL可简化为select(func.generate_series(0,23).label('hour'))，MySQL 8+可使用递归CTE生成序列。
• 类型匹配：用cast(Integer)确保extract提取的小时与生成的小时序列类型一致，避免连接失败。
• 空值处理：coalesce函数会将左连接产生的NULL（无数据的小时）替换为0，无需Python层面补全。
• 日期筛选：用func.date(Shapes.timestamp)筛选特定日期，比between更精准，不会遗漏或超出目标日期范围。

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