请求协助将指定SQL语句转换为SQLAlchemy可执行查询

没问题，我来帮你把这条PostgreSQL专属的SQL转换成对应的SQLAlchemy查询语句。先拆解下原SQL的核心逻辑：它要获取每个应用对应的最新版本记录（按created时间倒序取第一条），关联应用表的基础信息，最后筛选出状态为DECLINED的版本，再按应用ID和版本创建时间排序。

第一步：假设你的模型结构 #

首先假设你已经定义了对应的SQLAlchemy模型（如果你的字段名/表名有差异，直接替换成自己的即可）：

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class Applications(Base):
    __tablename__ = 'applications'
    id = Column(Integer, primary_key=True)
    developer_id = Column(Integer)
    name = Column(String)
    description = Column(Text)
    created = Column(DateTime)
    updated = Column(DateTime)
    versions = relationship("ApplicationsVersions", back_populates="application")

class ApplicationsVersions(Base):
    __tablename__ = 'applications_versions'
    id = Column(Integer, primary_key=True)
    application_id = Column(Integer, ForeignKey('applications.id'))
    version = Column(String)
    path = Column(String)
    status = Column(String)
    release_notes = Column(Text)
    created = Column(DateTime)
    updated = Column(DateTime)
    application = relationship("Applications", back_populates="versions")

第二步：构造子查询（对应原SQL的anon_1） #

原SQL里的子查询用了PostgreSQL的DISTINCT ON特性来获取每个应用的最新版本，在SQLAlchemy 1.4+版本中可以直接用distinct_on()方法实现：

from sqlalchemy import select

# 构建子查询：获取每个应用的最新版本记录，关联应用表信息
subquery = (
    select(
        # 给版本表字段加别名，和原SQL保持一致
        ApplicationsVersions.id.label("applications_versions_id"),
        ApplicationsVersions.application_id.label("applications_versions_application_id"),
        ApplicationsVersions.version.label("applications_versions_version"),
        ApplicationsVersions.path.label("applications_versions_path"),
        ApplicationsVersions.status.label("applications_versions_status"),
        ApplicationsVersions.release_notes.label("applications_versions_release_notes"),
        ApplicationsVersions.created.label("applications_versions_created"),
        ApplicationsVersions.updated.label("applications_versions_updated"),
        # 应用表字段加别名
        Applications.id.label("applications_id"),
        Applications.developer_id.label("applications_developer_id"),
        Applications.name.label("applications_name"),
        Applications.description.label("applications_description"),
        Applications.created.label("applications_created"),
        Applications.updated.label("applications_updated")
    )
    # 关联应用表和版本表
    .join(Applications, Applications.id == ApplicationsVersions.application_id)
    # PostgreSQL专属：按application_id去重，保留每组第一条
    .distinct_on(ApplicationsVersions.application_id)
    # 排序规则：先按application_id分组，再按created倒序取最新版本
    .order_by(ApplicationsVersions.application_id, ApplicationsVersions.created.desc())
).subquery()

第三步：构造外层查询 #

基于子查询筛选状态为DECLINED的记录，并返回需要的字段：

# 外层查询：筛选状态为DECLINED的记录，返回指定字段
final_query = (
    select(
        subquery.c.applications_versions_id.label("versions_id"),
        subquery.c.applications_versions_application_id.label("versions_application_id"),
        subquery.c.applications_versions_version.label("version"),
        subquery.c.applications_versions_status.label("status"),
        subquery.c.applications_id.label("applications_id"),
        subquery.c.applications_developer_id.label("developer_id"),
        subquery.c.applications_name.label("name"),
        subquery.c.applications_description.label("description")
    )
    # 筛选状态为DECLINED的版本
    .where(subquery.c.applications_versions_status == 'DECLINED')
    # 最终排序规则
    .order_by(subquery.c.applications_versions_application_id, subquery.c.applications_versions_created.desc())
)

第四步：执行查询 #

用SQLAlchemy的Session执行这个查询即可：

# 假设你已经创建了session对象
results = session.execute(final_query).all()

# 如果需要转换成ORM对象（比如字典），可以这样处理
result_dicts = [row._asdict() for row in results]

兼容旧版本SQLAlchemy（1.3及以下） #

如果你的SQLAlchemy版本低于1.4，不支持distinct_on()，可以用窗口函数row_number()来实现相同逻辑：

from sqlalchemy import func, over

# 先给每条版本记录按应用分组编号，最新的版本编号为1
ranked_subquery = (
    select(
        ApplicationsVersions,
        Applications,
        func.row_number().over(
            partition_by=ApplicationsVersions.application_id,
            order_by=ApplicationsVersions.created.desc()
        ).label("rn")
    )
    .join(Applications, Applications.id == ApplicationsVersions.application_id)
).subquery()

# 再筛选编号为1的记录（即每个应用的最新版本），后续逻辑和之前一致
final_query = (
    select(
        ranked_subquery.c.applications_versions_id.label("versions_id"),
        ranked_subquery.c.applications_versions_application_id.label("versions_application_id"),
        ranked_subquery.c.applications_versions_version.label("version"),
        ranked_subquery.c.applications_versions_status.label("status"),
        ranked_subquery.c.applications_id.label("applications_id"),
        ranked_subquery.c.applications_developer_id.label("developer_id"),
        ranked_subquery.c.applications_name.label("name"),
        ranked_subquery.c.applications_description.label("description")
    )
    .where(ranked_subquery.c.rn == 1)
    .where(ranked_subquery.c.applications_versions_status == 'DECLINED')
    .order_by(ranked_subquery.c.applications_versions_application_id, ranked_subquery.c.applications_versions_created.desc())
)

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