FastAPI+SQLAlchemy：查询父记录时仅返回最新子记录问题

先假设你的模型结构大致如下（如果和实际有出入，调整对应字段即可）：

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

Base = declarative_base()

class Sample(Base):
    __tablename__ = "samples"
    id = Column(Integer, primary_key=True, index=True)
    sample_code = Column(String, unique=True, index=True)  # 示例字段
    location_logs = relationship("LocationLog", back_populates="sample")

class LocationLog(Base):
    __tablename__ = "location_logs"
    id = Column(Integer, primary_key=True, index=True)
    sample_id = Column(Integer, ForeignKey("samples.id"))
    location = Column(String)
    time_created = Column(DateTime)
    sample = relationship("Sample", back_populates="location_logs")

你之前的问题可能出在这 #

1. 子查询未同时关联sample_id和time_created两个条件，导致过滤失效
2. 直接依赖ORM默认的relationship返回所有子记录，没通过查询语句做过滤
3. 分组或排序逻辑错误，没精准定位到每个样本的最新日志

两种可行实现方式 #

方式一：子查询分组取最大时间（适合简单场景） #

先通过子查询获取每个样本对应的最新日志时间，再关联主表筛选出对应记录：

from sqlalchemy import select, func
from sqlalchemy.orm import Session

def get_samples_with_latest_log(db: Session):
    # 子查询：按样本分组，取每个组的最新日志时间
    latest_time_subq = (
        select(
            LocationLog.sample_id,
            func.max(LocationLog.time_created).label("latest_time")
        )
        .group_by(LocationLog.sample_id)
        .subquery()
    )

    # 关联样本表、日志表和子查询，筛选出最新日志
    query = (
        select(Sample, LocationLog)
        .join(LocationLog, Sample.id == LocationLog.sample_id)
        .join(
            latest_time_subq,
            (LocationLog.sample_id == latest_time_subq.c.sample_id) &
            (LocationLog.time_created == latest_time_subq.c.latest_time)
        )
    )

    # 整理结果：给每个Sample对象绑定最新日志
    results = db.execute(query).all()
    for sample, latest_log in results:
        sample.latest_location_log = latest_log
    return [item[0] for item in results]

方式二：窗口函数排序取首条（适合复杂排序/去重场景） #

利用PostgreSQL的窗口函数ROW_NUMBER()，给每个样本的日志按时间倒序排名，取排名第1的记录：

from sqlalchemy import select, func, over
from sqlalchemy.orm import Session

def get_samples_with_latest_log(db: Session):
    # 子查询：给每个样本的日志按时间倒序排名
    ranked_logs_subq = (
        select(
            LocationLog,
            func.row_number().over(
                partition_by=LocationLog.sample_id,
                order_by=LocationLog.time_created.desc()
            ).label("log_rank")
        )
        .subquery()
    )

    # 关联样本表和排名后的日志表，取排名第1的记录
    query = (
        select(Sample, ranked_logs_subq)
        .join(ranked_logs_subq, Sample.id == ranked_logs_subq.c.sample_id)
        .where(ranked_logs_subq.c.log_rank == 1)
    )

    results = db.execute(query).all()
    for sample, latest_log in results:
        sample.latest_location_log = latest_log
    return [item[0] for item in results]

可选：在模型中直接定义关联（单查询场景友好） #

如果需要在查询单个Sample时直接获取最新日志，可以在Sample模型中新增一个只读关联：

class Sample(Base):
    __tablename__ = "samples"
    id = Column(Integer, primary_key=True, index=True)
    sample_code = Column(String, unique=True, index=True)
    location_logs = relationship("LocationLog", back_populates="sample")
    
    # 新增：关联最新的位置日志（uselist=False表示只返回一条）
    latest_location_log = relationship(
        "LocationLog",
        primaryjoin="""and_(
            Sample.id == LocationLog.sample_id,
            LocationLog.time_created == (
                select(func.max(LocationLog.time_created))
                .where(LocationLog.sample_id == Sample.id)
            )
        )""",
        uselist=False,
        viewonly=True
    )

之后查询db.query(Sample).filter(...).first()时，直接通过sample.latest_location_log就能拿到最新日志，但批量查询时建议用前两种方式，避免多次子查询影响性能。

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