SQLAlchemy三表联查获取app与env组合下最新deploy_date对应数据

模型定义 #

现有三个SQLAlchemy模型（无关字段已省略）：

class Application():
  id: int = db.Column(db.Integer, primary_key=True, autoincrement=True)
  name: str = db.Column(db.String(10), unique=True, index=True, nullable=False)

  appmappings: List[AppEnvMapping] = db.orm.relationship("AppEnvMapping", order_by=AppEnvMapping.deploy_date.desc(), back_populates='application', lazy='joined')

class Environment():
  id: int = db.Column(db.Integer, primary_key=True, autoincrement=True)
  name: str = db.Column(db.String(10), unique=True, index=True, nullable=False)

  envmappings: List[AppEnvMapping] = db.orm.relationship("AppEnvMapping", order_by=AppEnvMapping.deploy_date.desc(), back_populates='environment', lazy='joined')

class AppEnvMapping():
  id: int = db.Column(db.Integer, primary_key=True, autoincrement=True)
  app_id: int = db.Column(db.Integer, db.ForeignKey('applications.id'), index=True, nullable=False)
  env_id: int = db.Column(db.Integer, db.ForeignKey('environments.id'), index=True, nullable=False)
  deploy_date: datetime.datetime = db.Column(db.DateTime, index=True, nullable=False)

  application = db.orm.relationship('Application', back_populates='appmappings', lazy='joined')
  environment = db.orm.relationship('Environment', back_populates='appmappings', lazy='joined')

需求说明 #

查询每个app_id和env_id组合下deploy_date最新的AppEnvMapping记录，同时返回对应的Application和Environment关联数据。
示例AppEnvMapping表数据：

id | app_id | env_id | deploy_date
1  | 1      | 1      | 01-01-2021
2  | 1      | 1      | 02-01-2021
3  | 1      | 2      | 01-10-2021
4  | 2      | 1      | 02-04-2021
5  | 2      | 1      | 04-15-2021

期望返回结果：

id | app_id | env_id | deploy_date
2  | 1      | 1      | 02-01-2021
3  | 1      | 2      | 01-10-2021
5  | 2      | 1      | 04-15-2021

现有实现问题 #

原有子查询逻辑存在错误：分组查询时直接返回未加入分组条件的id字段，得到的是分组内随机值，导致关联条件失效，最终返回全量数据。同时代码存在拼写错误、括号未闭合等语法问题。

方案1：修正子查询关联逻辑（兼容所有SQLAlchemy版本） #

先构造子查询仅返回每个app_id+env_id组合的最大部署日期，再关联三个匹配字段拿到对应记录：

from sqlalchemy import func, and_

# 构造子查询：取每个应用+环境组合的最大部署日期
max_deploy_subq = session.query(
    AppEnvMapping.app_id,
    AppEnvMapping.env_id,
    func.max(AppEnvMapping.deploy_date).label("max_deploy_date")
).group_by(
    AppEnvMapping.app_id,
    AppEnvMapping.env_id
).subquery("max_deploy")

# 关联查询获取符合条件的记录，预加载关联数据避免N+1查询
result = session.query(AppEnvMapping).join(
    max_deploy_subq,
    and_(
        AppEnvMapping.app_id == max_deploy_subq.c.app_id,
        AppEnvMapping.env_id == max_deploy_subq.c.env_id,
        AppEnvMapping.deploy_date == max_deploy_subq.c.max_deploy_date
    )
).options(
    db.orm.joinedload(AppEnvMapping.application),
    db.orm.joinedload(AppEnvMapping.environment)
).all()

返回结果中的每个AppEnvMapping对象可直接通过.application、.environment属性拿到关联的应用、环境数据。

方案2：窗口函数实现（SQLAlchemy 1.4+ 推荐） #

使用ROW_NUMBER窗口函数分区排序后取每组第一条，逻辑更简洁：

from sqlalchemy import func, over

# 按应用+环境分区，按部署日期倒序排序，标注每行的分区内序号
row_num = over(
    func.row_number(),
    partition_by=[AppEnvMapping.app_id, AppEnvMapping.env_id],
    order_by=AppEnvMapping.deploy_date.desc()
).label("row_num")

subq = session.query(AppEnvMapping.id, row_num).subquery()

# 仅取每个分区序号为1的记录（即最新部署的记录）
result = session.query(AppEnvMapping).join(
    subq,
    and_(AppEnvMapping.id == subq.c.id, subq.c.row_num == 1)
).options(
    db.orm.joinedload(AppEnvMapping.application),
    db.orm.joinedload(AppEnvMapping.environment)
).all()

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