FastAPI+SQLAlchemy按会话聚合消息：关联查询结果优化问题

问题背景 #

使用FastAPI + SQLAlchemy开发社交媒体消息系统，需要通过user_id获取按会话分组的所有消息。当前查询返回的结果中，同一会话的消息被拆分为多个独立字典，期望将同一会话的所有消息合并到同一个字典中。

当前响应示例 #

[
  {
    "Conversation": {
      "id_post": 2,
      "id_conversation": 1
    },
    "Message": {
      "id_conversation": 1,
      "id_post": 2,
      "id_receiver": 2,
      "date_insert": "2023-12-12T16:38:35.703764",
      "id_sender": 1,
      "id_message": 1,
      "message_text": "Hello",
      "is_read": false
    }
  },
  {
    "Conversation": {
      "id_post": 2,
      "id_conversation": 1
    },
    "Message": {
      "id_conversation": 1,
      "id_post": 2,
      "id_receiver": 2,
      "date_insert": "2023-12-12T16:39:10.112846",
      "id_sender": 1,
      "id_message": 2,
      "message_text": "are u here ?",
      "is_read": false
    }
  },
  {
    "Conversation": {
      "id_post": 1,
      "id_conversation": 3
    },
    "Message": {
      "id_conversation": 3,
      "id_post": 1,
      "id_receiver": 2,
      "date_insert": "2023-12-12T16:40:53.849667",
      "id_sender": 3,
      "id_message": 4,
      "message_text": "Hey man",
      "is_read": false
    }
  }
]

模型代码（models.py） #

class Conversation(Base):
    __tablename__ = "conversations"

    id_conversation = Column(Integer, primary_key=True, index=True)
    id_post = Column(Integer, ForeignKey("posts.id_post", ondelete="cascade"),nullable=False)
    messages = relationship('Message', back_populates='conversation')


    class Config:
        orm_mode = True

class Message(Base):
    __tablename__ = "messages"

    id_message =  Column(Integer, primary_key=True, index=True)
    id_conversation = Column(Integer, ForeignKey("conversations.id_conversation", ondelete="cascade"),nullable=False)
    id_post = Column(Integer, ForeignKey("posts.id_post", ondelete="cascade"),nullable=False)
    id_sender = Column(Integer, ForeignKey("users.id_user", ondelete="cascade"),nullable=False)
    id_receiver = Column(Integer, ForeignKey("users.id_user", ondelete="cascade"),nullable=False)
    message_text = Column(String(1000), nullable=False)
    date_insert = Column(DateTime, nullable=False, default=func.now())
    is_read = Column(Boolean, nullable=False, default=False)
    conversation = relationship('Conversation', back_populates='messages')


    class Config:
        orm_mode = True

当前查询代码 #

def get_all_messages(db: Session, id_user: int):
    all_messages = db.query(models.Conversation, models.Message
                            ).join(models.Message, models.Conversation.id_conversation == models.Message.id_conversation
                                   ).filter(or_(models.Message.id_sender == id_user, 
                                                models.Message.id_receiver == id_user)
                                            ).all()

遇到的错误 #

尝试使用.group_by(models.Conversation.id_conversation)或.group_by(models.Conversation)时，报错：column "messages.id_message" must appear in the GROUP BY

解决方案 #

方法1：利用SQLAlchemy关系直接查询（推荐） #

因为已在Conversation模型中定义了与Message的关联关系，可以直接查询用户参与的会话并预加载关联消息，无需手动JOIN和分组。

修改后的查询代码：

from sqlalchemy.orm import joinedload

def get_all_messages(db: Session, id_user: int):
    # 查询用户参与的会话并预加载关联消息
    conversations = db.query(models.Conversation)\
        .join(models.Message)\
        .filter(or_(models.Message.id_sender == id_user, models.Message.id_receiver == id_user))\
        .options(joinedload(models.Conversation.messages))\
        .distinct()\
        .all()
    
    # 转换为目标格式
    result = []
    for conv in conversations:
        result.append({
            "Conversation": {
                "id_conversation": conv.id_conversation,
                "id_post": conv.id_post
            },
            "Messages": [
                {
                    "id_message": msg.id_message,
                    "id_conversation": msg.id_conversation,
                    "id_post": msg.id_post,
                    "id_sender": msg.id_sender,
                    "id_receiver": msg.id_receiver,
                    "message_text": msg.message_text,
                    "date_insert": msg.date_insert.isoformat(),
                    "is_read": msg.is_read
                } for msg in conv.messages
                # 可选：过滤当前用户参与的消息（避免会话包含无关消息）
                if msg.id_sender == id_user or msg.id_receiver == id_user
            ]
        })
    return result

说明：

• joinedload预加载关联消息，避免N+1查询性能问题
• distinct()确保每个会话仅返回一次
• 最终将ORM对象转换为目标字典结构，同一会话的消息统一放在Messages列表中

方法2：Python层面手动分组 #

如果坚持使用原查询方式，可以在获取结果后，通过Python代码按会话ID分组：

from collections import defaultdict

def get_all_messages(db: Session, id_user: int):
    all_messages = db.query(models.Conversation, models.Message
                            ).join(models.Message, models.Conversation.id_conversation == models.Message.id_conversation
                                   ).filter(or_(models.Message.id_sender == id_user, 
                                                models.Message.id_receiver == id_user)
                                            ).all()
    
    # 按会话ID分组整理数据
    grouped = defaultdict(lambda: {"Conversation": None, "Messages": []})
    for conv, msg in all_messages:
        conv_id = conv.id_conversation
        if grouped[conv_id]["Conversation"] is None:
            grouped[conv_id]["Conversation"] = {
                "id_conversation": conv.id_conversation,
                "id_post": conv.id_post
            }
        grouped[conv_id]["Messages"].append({
            "id_message": msg.id_message,
            "id_conversation": msg.id_conversation,
            "id_post": msg.id_post,
            "id_sender": msg.id_sender,
            "id_receiver": msg.id_receiver,
            "message_text": msg.message_text,
            "date_insert": msg.date_insert.isoformat(),
            "is_read": msg.is_read
        })
    
    # 转换为列表格式返回
    return list(grouped.values())

关于GROUP BY报错的原因 #

SQL的GROUP BY要求SELECT中的所有非聚合列必须出现在GROUP BY子句中。原查询同时选择了Conversation和Message的所有列，而messages.id_message是唯一值无法被聚合，因此数据库要求它必须加入GROUP BY，但这会导致每个消息单独成组，无法实现会话级别的合并。因此不适合用SQL的GROUP BY实现该需求。

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