Spliting to private functions and fixing small issues

Author: ljavorsk

common/models.py

@@ -5,6 +5,7 @@
 and components to ensure consistency and type safety.
 """
 
+from datetime import datetime
 from typing import Optional, Dict, Any, Union
 from pydantic import BaseModel, Field
 from pathlib import Path
@@ -299,8 +300,8 @@ class CommentReply(BaseModel):
 
     author: str = Field(description="Username of the reply author")
     message: str = Field(description="The reply message text")
-    created_at: str = Field(
-        description="ISO 8601 timestamp when reply was created"
+    created_at: datetime = Field(
+        description="Timestamp when reply was created"
     )
 
 
@@ -309,8 +310,8 @@ class MergeRequestComment(BaseModel):
 
     author: str = Field(description="Username of the comment author")
     message: str = Field(description="The comment message text")
-    created_at: str = Field(
-        description="ISO 8601 timestamp when comment was created"
+    created_at: datetime = Field(
+        description="Timestamp when comment was created"
     )
     file_path: str = Field(
         default="",
@@ -338,4 +339,4 @@ class MergeRequestComment(BaseModel):
         default_factory=list,
         description="List of replies to this comment in the thread, "
         "ordered chronologically"
-    )
+    )

mcp_server/gitlab_tools.py

@@ -19,6 +19,10 @@
 
 logger = logging.getLogger(__name__)
 
+# GitLab access levels: Guest (10), Reporter (20), Developer (30),
+# Maintainer (40), Owner (50)
+DEVELOPER_ACCESS_LEVEL = 30
+
 
 def _get_authenticated_url(repository_url: str) -> str:
     """
@@ -388,105 +392,122 @@ def get_latest_pipeline_jobs():
         raise ToolError(f"Failed to get failed jobs from merge request: {e}") from e
 
 
+def _get_authorized_member_ids(project) -> set[int]:
+    """
+    Fetch all project members and return a set of IDs for members
+    with Developer role or higher. This avoids N+1 API calls.
+    """
+    try:
+        members = project.members_all.list(get_all=True)
+        return {
+            member.id for member in members
+            if member.access_level >= DEVELOPER_ACCESS_LEVEL
+        }
+    except Exception as e:
+        logger.warning(f"Failed to fetch project members: {e}")
+        return set()
+
+
+def _extract_position_info(note: dict) -> tuple[str, int | None, str]:
+    """Extract file path, line number, and line type from a note's position."""
+    position = note.get("position")
+    if not position:
+        return "", None, ""
+
+    file_path = position.get("new_path", "") or position.get("old_path", "")
+    new_line = position.get("new_line")
+    old_line = position.get("old_line")
+
+    if new_line and old_line:
+        return file_path, new_line, "unchanged"
+    elif new_line:
+        return file_path, new_line, "new"
+    elif old_line:
+        return file_path, old_line, "old"
+
+    return file_path, None, ""
+
+
+def _process_reply(
+    authorized_member_ids: set[int], note: dict
+) -> CommentReply | None:
+    """Process a reply note and return CommentReply if author is authorized."""
+    if note.get("system", False):
+        return None
+
+    try:
+        author = note.get("author", {})
+        author_id = author.get("id")
+        if not author_id or author_id not in authorized_member_ids:
+            return None
+
+        return CommentReply(
+            author=author.get("username", "unknown"),
+            message=note.get("body", ""),
+            created_at=note.get("created_at", ""),
+        )
+    except Exception as e:
+        logger.warning(f"Failed to process reply note: {e}")
+        return None
+
+
 async def get_authorized_comments_from_merge_request(
     merge_request_url: Annotated[str, Field(description="URL of the merge request")],
 ) -> list[MergeRequestComment]:
     """
     Gets all comments from a merge request, filtered to only include
     comments from authorized members with Developer role or higher.
-    Access levels: Guest (10), Reporter (20), Developer (30),
-    Maintainer (40), Owner (50).
     """
     try:
         mr = await _get_merge_request_from_url(merge_request_url)
 
         def get_authorized_comments():
-            # Get discussions instead of notes to capture thread context
             discussions = mr._raw_pr.discussions.list(get_all=True)
             project = mr.target_project.gitlab_repo
 
-            logger.info(f"Found {len(discussions)} discussions in MR")
+            # Fetch all authorized members once to avoid N+1 API calls
+            authorized_member_ids = _get_authorized_member_ids(project)
 
             authorized_comments = []
             for discussion in discussions:
-                discussion_id = discussion.id
-                notes = discussion.attributes.get("notes", [])
-
-                if not notes:
-                    continue
-
-                # Process first note (thread starter)
-                first_note = notes[0]
-
-                # Skip system notes (like "added 1 commit")
-                if first_note.get("system", False): continue
-
-                first_author_id = first_note["author"]["id"]
-                first_author_username = first_note["author"]["username"]
-
                 try:
-                    first_member = project.members_all.get(first_author_id)
-
-                    # Only process thread if starter is Developer+
-                    if first_member.access_level >= 30:
-                        file_path = ""
-                        line_number = None
-                        line_type = ""
-
-                        position = first_note.get("position")
-
-                        if position:
-                            file_path = (position.get("new_path", "") or position.get("old_path", ""))
-                            new_line = position.get("new_line")
-                            old_line = position.get("old_line")
-
-                            if new_line and old_line:
-                                # Both present = unchanged/context line
-                                line_number = new_line
-                                line_type = "unchanged"
-                            elif new_line:
-                                line_number = new_line
-                                line_type = "new"
-                            elif old_line:
-                                line_number = old_line
-                                line_type = "old"
-
-                        replies = []
-                        for reply_note in notes[1:]:
-                            if reply_note.get("system", False): continue
-
-                            reply_author_id = reply_note["author"]["id"]
-                            reply_author_username = reply_note["author"]["username"]
-
-                            try:
-                                reply_member = project.members_all.get(reply_author_id)
-
-                                if reply_member.access_level >= 30:
-                                    replies.append(
-                                        CommentReply(
-                                            author=reply_author_username,
-                                            message=reply_note["body"],
-                                            created_at=reply_note["created_at"],
-                                        )
-                                    )
-                            except Exception as e:
-                                # Reply author is not an authorized member
-                                continue
-
-                        authorized_comments.append(
-                            MergeRequestComment(
-                                author=first_author_username,
-                                message=first_note["body"],
-                                created_at=first_note["created_at"],
-                                file_path=file_path,
-                                line_number=line_number,
-                                line_type=line_type,
-                                discussion_id=discussion_id,
-                                replies=replies,
-                            )
+                    if not (notes := discussion.attributes.get("notes")):
+                        continue
+
+                    first_note = notes[0]
+
+                    # Skip system notes (e.g. commit added)
+                    if first_note.get("system"):
+                        continue
+
+                    author = first_note.get("author", {})
+                    author_id = author.get("id")
+                    if not author_id or author_id not in authorized_member_ids:
+                        continue
+
+                    file_path, line_number, line_type = (
+                        _extract_position_info(first_note)
+                    )
+
+                    replies = [
+                        reply for note in notes[1:]
+                        if (reply := _process_reply(authorized_member_ids, note)) is not None
+                    ]
+
+                    authorized_comments.append(
+                        MergeRequestComment(
+                            author=author.get("username", "unknown"),
+                            message=first_note.get("body", ""),
+                            created_at=first_note.get("created_at", ""),
+                            file_path=file_path,
+                            line_number=line_number,
+                            line_type=line_type,
+                            discussion_id=getattr(discussion, "id", ""),
+                            replies=replies,
                         )
+                    )
                 except Exception as e:
-                    # Thread starter is not an authorized member
+                    logger.warning(f"Failed to process discussion: {e}")
                     continue
 
             return authorized_comments