如何在Async SQLAlchemy中实现带Limit的selectinload？

问题1：获取每个聊天室的最后一条消息 #

你的代码通过selectinload(Chatroom.messages)加载了聊天室的全部消息，要只获取最后一条，最高效的方式是通过子查询关联直接查询最新消息，避免加载冗余数据。

假设你的Message模型包含chatroom_id（关联聊天室）、created_at（消息创建时间，用于判断最新）字段，示例代码如下：

from sqlalchemy import select, func, outerjoin, or_

# 子查询：获取每个聊天室的最新消息时间
latest_msg_subq = select(
    Message.chatroom_id,
    func.max(Message.created_at).label('latest_time')
).group_by(Message.chatroom_id).subquery()

# 查询聊天室，并关联对应的最新消息
chats = await self.session.execute(
    select(Chatroom, Message)
    .filter(
        or_(
            Chatroom.first_user == user_id,
            Chatroom.second_user == user_id
        )
    )
    .outerjoin(
        Message,
        (Chatroom.id == Message.chatroom_id) & 
        (Message.created_at == latest_msg_subq.c.latest_time)
    )
    .options(
        selectinload(Chatroom.first_user_link),
        selectinload(Chatroom.second_user_link)
    )
)

# 遍历结果，每个元组对应一个聊天室和它的最后一条消息
for chatroom, latest_message in chats.unique():
    print(f"聊天室 {chatroom.id} 的最后一条消息：{latest_message.content if latest_message else '无消息'}")

如果你的Message用自增ID判断新旧（ID越大越新），可以把func.max(Message.created_at)替换为func.max(Message.id)，关联条件对应改成Message.id == latest_msg_subq.c.latest_id。

问题2：懒加载的MissingGreenlet错误 #

异步SQLAlchemy中，默认的懒加载（lazy="select"）是同步实现的，在异步上下文直接访问会触发该错误，修复方式有两种：

• 配置异步兼容的懒加载：在模型定义时，将messages关联的lazy参数设为"selectin"或"joined"（这两种支持异步）：

  class Chatroom(Base):
      # ...其他字段定义
      messages = relationship("Message", lazy="selectin")
  

• 异步触发懒加载：访问懒加载属性时用await关键字：

  # 先获取单个聊天室
  chatroom = await self.session.get(Chatroom, chat_id)
  # 异步加载全部消息
  messages = await chatroom.messages
  # 取最后一条
  latest_msg = messages[-1] if messages else None
  

注意：第二种方式仍会加载所有消息，性能不如直接查询最新消息的方案，优先推荐前者。

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