使用SQLAlchemy获取每个父表对应的最新子表记录
解决方法
首先定义对应的SQLAlchemy模型:
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True) other = Column(String) children = relationship("Children", back_populates="parent") class Children(Base): __tablename__ = 'children' id = Column(Integer, primary_key=True) parent_id = Column(Integer, ForeignKey('parent.id')) time_created = Column(DateTime) parent = relationship("Parent", back_populates="children")
方法一:子查询关联筛选
先通过子查询获取每个parent_id对应的最新time_created,再关联Children表匹配出对应记录:
from sqlalchemy import select, func, and_ # 子查询:分组计算每个父级的最新创建时间 subquery = select( Children.parent_id, func.max(Children.time_created).label('max_time') ).group_by(Children.parent_id).subquery() # 主查询:关联子查询,筛选出每个父级对应最新时间的子记录 query = select(Children).join( subquery, and_( Children.parent_id == subquery.c.parent_id, Children.time_created == subquery.c.max_time ) ) # 执行查询(需确保已创建SQLAlchemy Session) latest_children = session.execute(query).scalars().all()
方法二:窗口函数筛选
利用ROW_NUMBER()窗口函数按父级分组、创建时间倒序排序,取每组排名第一的记录:
from sqlalchemy import select, func, over # 定义窗口函数:按parent_id分组,time_created降序生成行号 row_number = over( func.row_number(), partition_by=Children.parent_id, order_by=Children.time_created.desc() ).label('row_num') # 子查询:为每条子记录添加行号字段 subquery = select(Children, row_number).subquery() # 主查询:筛选出每组行号为1的最新记录 query = select(subquery).where(subquery.c.row_num == 1) # 执行查询 latest_children = session.execute(query).scalars().all()
上述两种方法均可获取每个父级对应的最新Children记录(即ID为2和4的记录)。
内容的提问来源于stack exchange,提问作者Loic
相关产品推荐
相关产品推荐

