Function get_conversations

server/services/assistant_database.py:155–198

Get all conversations for a project with message counts. Uses a subquery for message_count to avoid N+1 query problem.

(project_dir: Path, project_name: str)

def get_conversations(project_dir: Path, project_name: str) -> list[dict]:
    """Get all conversations for a project with message counts.

    Uses a subquery for message_count to avoid N+1 query problem.
    """
    session = get_session(project_dir)
    try:
        # Subquery to count messages per conversation (avoids N+1 query)
        message_count_subquery = (
            session.query(
                ConversationMessage.conversation_id,
                func.count(ConversationMessage.id).label("message_count")
            )
            .group_by(ConversationMessage.conversation_id)
            .subquery()
        )

        # Join conversation with message counts
        conversations = (
            session.query(
                Conversation,
                func.coalesce(message_count_subquery.c.message_count, 0).label("message_count")
            )
            .outerjoin(
                message_count_subquery,
                Conversation.id == message_count_subquery.c.conversation_id
            )
            .filter(Conversation.project_name == project_name)
            .order_by(Conversation.updated_at.desc())
            .all()
        )
        return [
            {
                "id": c.Conversation.id,
                "project_name": c.Conversation.project_name,
                "title": c.Conversation.title,
                "created_at": c.Conversation.created_at.isoformat() if c.Conversation.created_at else None,
                "updated_at": c.Conversation.updated_at.isoformat() if c.Conversation.updated_at else None,
                "message_count": c.message_count,
            }
            for c in conversations
        ]
    finally:
        session.close()


def get_conversation(project_dir: Path, conversation_id: int) -> Optional[dict]: