SQLAlchemy按日期分组聚合计数：如何填充缺失日期

碰到这种日期缺失的情况，核心思路就是先生成一个覆盖查询时间范围的连续日期序列，再把原有的分组统计结果和这个序列做左连接，对缺失日期的统计值填充默认值（比如0）就行。下面分两种常用场景给你具体代码示例：

一、数据库层面生成日期序列（推荐，效率更高） #

不同数据库生成连续日期的语法略有差异，这里给你几个主流数据库的实现方式：

1. PostgreSQL 示例 #

利用PostgreSQL自带的generate_series函数快速生成连续日期：

from sqlalchemy import func, select
from your_app.models import Pomo
from sqlalchemy.orm import Session

def get_filled_pomo_data(session: Session):
    # 第一步：获取数据的时间范围（最小/最大日期）
    date_range = session.query(
        func.date(func.min(Pomo.created_at)).label("min_date"),
        func.date(func.max(Pomo.created_at)).label("max_date")
    ).first()
    if not date_range:
        return []
    min_date, max_date = date_range.min_date, date_range.max_date

    # 第二步：生成连续日期序列的CTE
    dates_cte = select(
        func.generate_series(min_date, max_date, interval="1 day").label("date")
    ).cte("dates")

    # 第三步：原分组统计查询的CTE
    pomo_stats_cte = select(
        func.date(Pomo.created_at).label("date"),
        func.count(Pomo.id).label("pomo_count")
    ).group_by(func.date(Pomo.created_at)).cte("pomo_stats")

    # 第四步：左连接并填充缺失值（用coalesce把NULL替换成0）
    result = session.query(
        dates_cte.c.date,
        func.coalesce(pomo_stats_cte.c.pomo_count, 0).label("pomo_count")
    ).outerjoin(
        pomo_stats_cte,
        dates_cte.c.date == pomo_stats_cte.c.date
    ).order_by(dates_cte.c.date).all()

    return result

2. MySQL 8.0+ 示例 #

MySQL 8.0及以上支持递归CTE，用它生成连续日期：

from sqlalchemy import func, select
from your_app.models import Pomo
from sqlalchemy.orm import Session

def get_filled_pomo_data(session: Session):
    date_range = session.query(
        func.date(func.min(Pomo.created_at)).label("min_date"),
        func.date(func.max(Pomo.created_at)).label("max_date")
    ).first()
    if not date_range:
        return []
    min_date, max_date = date_range.min_date, date_range.max_date

    # 递归CTE生成连续日期
    dates_cte = select(
        func.date_add(min_date, func.interval(func.t.n, "day")).label("date")
    ).select_from(
        select(func.t.n).select_from(
            select(func.sequence(0, func.datediff(max_date, min_date)).label("n")).cte("t")
        )
    ).cte("dates")

    # 原分组统计
    pomo_stats_cte = select(
        func.date(Pomo.created_at).label("date"),
        func.count(Pomo.id).label("pomo_count")
    ).group_by(func.date(Pomo.created_at)).cte("pomo_stats")

    # 左连接填充
    result = session.query(
        dates_cte.c.date,
        func.coalesce(pomo_stats_cte.c.pomo_count, 0).label("pomo_count")
    ).outerjoin(
        pomo_stats_cte,
        dates_cte.c.date == pomo_stats_cte.c.date
    ).order_by(dates_cte.c.date).all()

    return result

3. SQLite 示例 #

SQLite同样支持递归CTE，结合julianday函数计算日期差：

from sqlalchemy import func, select
from your_app.models import Pomo
from sqlalchemy.orm import Session

def get_filled_pomo_data(session: Session):
    date_range = session.query(
        func.date(func.min(Pomo.created_at)).label("min_date"),
        func.date(func.max(Pomo.created_at)).label("max_date")
    ).first()
    if not date_range:
        return []
    min_date, max_date = date_range.min_date, date_range.max_date

    # 递归CTE生成连续日期
    dates_cte = select(
        func.date(min_date, f"+{func.t.n} day").label("date")
    ).select_from(
        select(func.t.n).select_from(
            select(func.generate_series(0, func.julianday(max_date) - func.julianday(min_date)).label("n")).cte("t")
        )
    ).cte("dates")

    # 原分组统计
    pomo_stats_cte = select(
        func.date(Pomo.created_at).label("date"),
        func.count(Pomo.id).label("pomo_count")
    ).group_by(func.date(Pomo.created_at)).cte("pomo_stats")

    # 左连接填充
    result = session.query(
        dates_cte.c.date,
        func.coalesce(pomo_stats_cte.c.pomo_count, 0).label("pomo_count")
    ).outerjoin(
        pomo_stats_cte,
        dates_cte.c.date == pomo_stats_cte.c.date
    ).order_by(dates_cte.c.date).all()

    return result

二、Python层面生成日期序列（兼容所有数据库） #

如果你的数据库不支持复杂CTE或者你想统一逻辑，可以在Python端生成连续日期，再和查询结果匹配填充：

from datetime import timedelta, date
from sqlalchemy import func, select
from your_app.models import Pomo
from sqlalchemy.orm import Session

# 生成连续日期的工具函数
def generate_date_range(start_date: date, end_date: date):
    current_date = start_date
    while current_date <= end_date:
        yield current_date
        current_date += timedelta(days=1)

def get_filled_pomo_data(session: Session):
    # 获取时间范围
    date_range = session.query(
        func.date(func.min(Pomo.created_at)).label("min_date"),
        func.date(func.max(Pomo.created_at)).label("max_date")
    ).first()
    if not date_range:
        return []
    min_date, max_date = date_range.min_date, date_range.max_date

    # 生成所有连续日期
    all_dates = list(generate_date_range(min_date, max_date))

    # 查询原分组数据并转成字典（方便快速查找）
    pomo_stats = {
        row.date: row.pomo_count
        for row in session.query(
            func.date(Pomo.created_at).label("date"),
            func.count(Pomo.id).label("pomo_count")
        ).group_by(func.date(Pomo.created_at)).all()
    }

    # 填充缺失日期的统计值
    filled_result = [
        (d, pomo_stats.get(d, 0))
        for d in all_dates
    ]

    return filled_result

关键注意点 #

• 确保Pomo.created_at是datetime类型的字段，否则func.date()无法正确提取日期。
• func.coalesce()是SQL标准函数，用来将NULL值替换为指定的默认值（这里是0），保证每个日期都有统计值。
• 如果你的时间范围不是从数据的最小/最大日期开始，而是自定义的（比如最近30天），直接替换min_date和max_date为你的自定义日期即可。

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