获取rule_chains表当前生效实体的SQLAlchemy/原生SQL方案

现有rule_chains表，结构及数据如下：

需求为获取指定日期（如当日）的生效实体：

• 若日期为2023-03-01，需返回ID为1、2的实体；
• 若日期为2023-04-01，需返回ID为2、3的实体（payroll_type_id=1的生效实体为ID3）。

尝试编写的SQLAlchemy代码无法实现需求，请求可行的原生SQL或SQLAlchemy查询语句：

subquery = self.session.query(func.min(RuleChain.start_date).label("start_date"),
                              RuleChain.stuff_department_id,
                              RuleChain.payroll_type_id).group_by(
    RuleChain.stuff_department_id, RuleChain.payroll_type_id).having(
    func.max(RuleChain.start_date) >= self.date).subquery()

rule_chains: List[RuleChain] = self.session.query(RuleChain).filter(
    subquery.c.start_date == RuleChain.start_date,
    subquery.c.stuff_department_id == RuleChain.stuff_department_id,
    subquery.c.payroll_type_id == RuleChain.payroll_type_id
).all()

原生SQL实现 #

核心思路是先找出每个(payroll_type_id, stuff_department_id)组合下，不超过指定日期的最新生效日期，再关联原表取出对应记录：

SELECT rc.*
FROM rule_chains rc
INNER JOIN (
    SELECT 
        payroll_type_id,
        stuff_department_id,
        MAX(start_date) AS latest_start_date
    FROM rule_chains
    WHERE start_date <= '2023-04-01' -- 替换为你的指定日期
    GROUP BY payroll_type_id, stuff_department_id
) AS latest_rules
ON rc.payroll_type_id = latest_rules.payroll_type_id
AND rc.stuff_department_id = latest_rules.stuff_department_id
AND rc.start_date = latest_rules.latest_start_date;

SQLAlchemy实现 #

对应上述SQL的ORM写法：

from sqlalchemy import func
from typing import List

# 子查询：获取每个分组下不超过指定日期的最新生效日期
subquery = self.session.query(
    RuleChain.payroll_type_id,
    RuleChain.stuff_department_id,
    func.max(RuleChain.start_date).label("latest_start_date")
).filter(RuleChain.start_date <= self.date).group_by(
    RuleChain.payroll_type_id, RuleChain.stuff_department_id
).subquery()

# 关联查询获取最终生效实体
rule_chains: List[RuleChain] = self.session.query(RuleChain).join(
    subquery,
    (RuleChain.payroll_type_id == subquery.c.payroll_type_id) &
    (RuleChain.stuff_department_id == subquery.c.stuff_department_id) &
    (RuleChain.start_date == subquery.c.latest_start_date)
).all()

原代码问题说明 #

原代码使用func.min(RuleChain.start_date)获取分组最早日期，且having条件仅过滤存在晚于指定日期的记录，逻辑与需求不符——我们需要的是每个分组下不超过指定日期的最新生效记录，而非最早记录。

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