在SQLAlchemy中为购车信息表新增滚动年区间BMW购车数量列

嘿，我来帮你搞定这个需求！你要的是给每条购车记录计算过去365天（不包含当前记录日期）内BMW的交易总数，也就是示例里的nb_bmw_cars列对吧？结合你给的样本数据，我用SQLAlchemy给你写两种可行的实现方式，适配不同的数据库场景：

方法一：使用窗口函数（高效推荐，支持PostgreSQL/MySQL 8+等） #

如果你的数据库支持窗口函数，这种方法性能最优，不需要额外的子查询关联：

首先导入必要的模块（替换成你的模型类路径）：

from sqlalchemy import func, select, case
from sqlalchemy.orm import Session
from your_module import Purchase  # 你的购车记录模型

然后构造查询语句：

# 定义窗口计算逻辑：按客户分组，统计当前日期前365天到前1天的BMW数量
bmw_count_window = (
    func.count(
        case(
            [(Purchase.car_type == "BMW", 1)],  # 只统计BMW类型的记录
            else_=0
        )
    )
    .over(
        partition_by=Purchase.customer_id,  # 按客户单独统计（如果只针对特定客户可省略）
        order_by=Purchase.date,
        # 定义日期范围：当前日期减365天 到 当前日期减1天（开区间）
        range_=(
            func.date_sub(Purchase.date, func.interval(365, "day")),
            func.date_sub(Purchase.date, func.interval(1, "day"))
        )
    )
)

# 针对特定客户"xxxxxx"构造最终查询，同时处理0值为'-'
query = select(
    Purchase.customer_id,
    Purchase.date,
    Purchase.car_type,
    case(
        [(bmw_count_window == 0, "-")],  # 无历史记录时显示'-'
        else_=bmw_count_window
    ).label("nb_bmw_cars")
).where(Purchase.customer_id == "xxxxxx")

# 执行查询并获取结果
with Session(engine) as session:
    results = session.execute(query).all()
    # 遍历结果示例
    for row in results:
        print(f"客户ID: {row.customer_id}, 日期: {row.date}, 车型: {row.car_type}, 历史BMW数量: {row.nb_bmw_cars}")

方法二：使用子查询（兼容旧版数据库） #

如果你的数据库不支持窗口函数，可以用子查询关联的方式实现，兼容性更强：

from sqlalchemy import func, select, case
from sqlalchemy.orm import Session
from your_module import Purchase

# 给主表起别名，避免子查询字段冲突
main_purchase = Purchase.__table__.alias("main")
sub_purchase = Purchase.__table__.alias("sub")

# 构造子查询：统计当前主记录日期前365天内的BMW数量
bmw_count_subquery = select(
    func.count().label("bmw_count")
).select_from(sub_purchase)
.where(
    sub_purchase.c.customer_id == main_purchase.c.customer_id,  # 关联同一客户
    sub_purchase.c.car_type == "BMW",  # 筛选BMW车型
    # 日期范围：大于当前日期减365天，小于当前日期
    sub_purchase.c.date > func.date_sub(main_purchase.c.date, func.interval(365, "day")),
    sub_purchase.c.date < main_purchase.c.date
)

# 构造最终查询
query = select(
    main_purchase.c.customer_id,
    main_purchase.c.date,
    main_purchase.c.car_type,
    case(
        [(bmw_count_subquery.scalar_subquery() == 0, "-")],
        else_=bmw_count_subquery.scalar_subquery()
    ).label("nb_bmw_cars")
).where(main_purchase.c.customer_id == "xxxxxx")

# 执行查询
with Session(engine) as session:
    results = session.execute(query).all()
    # 处理结果
    for row in results:
        print(row)

关键逻辑说明 #

• 日期范围：我们用date_sub计算当前日期的前365天和前1天，确保只统计过去一年且不包含当天的BMW交易
• 空值处理：通过case函数把统计结果为0的情况转换成'-'，和你提供的示例格式一致
• 客户分组：两种方法都加入了customer_id的关联/分组，确保只统计当前客户的历史交易

备注：内容来源于stack exchange，提问作者redj