如何用SQLAlchemy 2查询用户的最新版本数据属性？

某个User对象拥有多个版本化的UserData数据集，查询时需要基于最新（最新创建）的UserData属性过滤用户，同时要兼容用户没有数据集的情况。期望实现类似：

# 返回对应用户（最新UserData的name是"user2"）
user = session.query(User).filter(User.latest_data.name == "user2").first()
# 返回None（最新UserData的name不是"user1"）
user = session.query(User).filter(User.latest_data.name == "user1").first()

解决方案：使用hybrid_property结合子查询 #

1. 导入必要模块 #

from sqlalchemy import select, func
from sqlalchemy.ext.hybrid import hybrid_property

2. 修改User类，添加latest_data属性 #

在User类中新增hybrid_property，同时实现Python层面访问和SQL层面查询的逻辑：

class User(Base):
    __tablename__ = 'user'

    id: Mapped[int] = mapped_column(primary_key=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    datas: Mapped[list[UserData]] = relationship(
        back_populates="user", order_by="desc(UserData.created_at)"
    )

    @hybrid_property
    def latest_data(self):
        # Python层面：直接取排序后的第一个数据集（无数据则返回None）
        return self.datas[0] if self.datas else None

    @latest_data.expression
    def latest_data(cls):
        # SQL层面：用窗口函数获取每个用户的最新UserData记录
        subquery = (
            select(
                UserData.id,
                func.row_number().over(
                    partition_by=UserData.user_id,
                    order_by=UserData.created_at.desc()
                ).label("rn")
            )
            .subquery()
        )
        # 关联回UserData表，筛选出每个用户编号为1的最新记录
        return select(UserData).where(
            (UserData.id == subquery.c.id) & (subquery.c.rn == 1)
        ).correlate(cls).scalar_subquery()

3. 验证查询逻辑 #

现在可以直接使用预期的查询方式：

# 查询最新数据集name为"user2"的用户
target_user = session.query(User).filter(User.latest_data.name == "user2").first()
# 查询最新数据集name为"user1"的用户（返回None）
non_exist_user = session.query(User).filter(User.latest_data.name == "user1").first()

4. 兼容无数据集的用户 #

如果需要筛选没有任何UserData的用户，可直接判断latest_data是否为None：

users_without_data = session.query(User).filter(User.latest_data.is_(None)).all()

逻辑说明 #

• Python层面：利用datas已按created_at倒序排列的特性，直接取第一个元素，简单高效。
• SQL层面：通过row_number()窗口函数为每个用户的数据集按创建时间倒序编号，取编号为1的记录，确保在数据库层面高效获取最新数据。

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