Track accurate PR cycle time by identifying first ready-for-review time

backend/analytics_server/mhq/exapi/github.py

@@ -1,7 +1,8 @@
 import contextlib
+import dataclasses
 from datetime import datetime
 from http import HTTPStatus
-from typing import Optional, Dict, Tuple, List
+from typing import Any, Optional, Dict, Tuple, List
 
 import requests
 from github import Github, UnknownObjectException
@@ -12,6 +13,16 @@
 from github.Repository import Repository as GithubRepository
 
 from mhq.exapi.models.github import GitHubContributor
+from mhq.exapi.types.timeline import (
+    GitHubReadyForReviewEventDict,
+    GitHubReviewEventFullDict,
+)
+from mhq.exapi.models.timeline import (
+    GitHubTimeline,
+    GitHubReadyForReviewEvent,
+    GitHubReviewEvent,
+    GitHubUser,
+)
 from mhq.utils.log import LOG
 
 PAGE_SIZE = 100
@@ -114,6 +125,230 @@ def get_pr_commits(self, pr: GithubPullRequest):
     def get_pr_reviews(self, pr: GithubPullRequest) -> GithubPaginatedList:
         return pr.get_reviews()
 
+    def get_pr_timeline(
+        self, org_login: str, repo_name: str, pr_number: int
+    ) -> List[GitHubTimeline]:
+        """
+        Fetches the timeline events for a pull request.
+
+        Args:
+            org_login: The organization login name
+            repo_name: The repository name
+            pr_number: The pull request number
+
+        Returns:
+            List of GitHub timeline events
+
+        Raises:
+            GithubException: If the API request fails
+        """
+        try:
+            timeline_events = self._fetch_all_timeline_events(
+                org_login, repo_name, pr_number
+            )
+            return self._process_timeline_events(timeline_events)
+        except GithubException as e:
+            LOG.error(
+                f"Failed to fetch PR timeline for {org_login}/{repo_name}#{pr_number}: {str(e)}"
+            )
+            raise
+
+    def _process_timeline_events(
+        self, timeline_events: List[Dict[str, Any]]
+    ) -> List[GitHubTimeline]:
+        """Process raw timeline events into model objects."""
+        result = []
+        for event in timeline_events:
+            if not self._is_supported_event(event):
+                continue
+
+            timeline_model = self._convert_to_timeline_model(event)
+            if timeline_model:
+                result.append(timeline_model)
+
+        return result
+
+    def _fetch_all_timeline_events(
+        self, org_login: str, repo_name: str, pr_number: int
+    ) -> List[Dict[str, Any]]:
+        """
+        Fetches all timeline events with pagination.
+
+        Args:
+            org_login: The organization login name
+            repo_name: The repository name
+            pr_number: The pull request number
+
+        Returns:
+            List of raw timeline events
+
+        Raises:
+            GithubException: If the API request fails
+        """
+        all_events = []
+        page = 1
+
+        while True:
+            github_url = f"{self.base_url}/repos/{org_login}/{repo_name}/issues/{pr_number}/timeline"
+            query_params = {"per_page": PAGE_SIZE, "page": page}
+
+            response = requests.get(
+                github_url,
+                headers={
+                    **self.headers,
+                    "Accept": "application/vnd.github.mockingbird-preview+json",
+                },
+                params=query_params,
+                timeout=15,
+            )
+
+            if response.status_code != HTTPStatus.OK:
+                raise GithubException(response.status_code, response.json())
+
+            data = response.json()
+            all_events.extend(data)
+
+            if len(data) < PAGE_SIZE:
+                break
+
+            page += 1
+
+        return all_events
+
+    def _is_supported_event(self, timeline_event: Dict[str, Any]) -> bool:
+        """
+        Checks if the event type is supported.
+
+        Args:
+            timeline_event: Raw timeline event data
+
+        Returns:
+            True if event type is supported, False otherwise
+        """
+        SUPPORTED_EVENTS = {"ready_for_review", "reviewed"}
+        return timeline_event.get("event") in SUPPORTED_EVENTS
+
+    def _convert_to_timeline_model(
+        self, timeline_event: Dict[str, Any]
+    ) -> Optional[GitHubTimeline]:
+        """
+        Converts raw event data into a typed model instance.
+
+        Args:
+            timeline_event: Raw timeline event data
+
+        Returns:
+            GitHubTimeline object or None if event type is not supported
+        """
+        event_type = timeline_event.get("event")
+
+        event_converters = {
+            "ready_for_review": self._create_ready_for_review_event,
+            "reviewed": self._create_review_event,
+        }
+
+        converter = event_converters.get(event_type)
+        if not converter:
+            return None
+
+        return converter(timeline_event)
+
+    def _create_ready_for_review_event(
+        self, event_data: Dict[str, Any]
+    ) -> GitHubReadyForReviewEvent:
+        """
+        Creates a ReadyForReview event from raw data.
+
+        Args:
+            event_data: Raw event data
+
+        Returns:
+            GitHubReadyForReviewEvent object
+        """
+        typed_dict = GitHubReadyForReviewEventDict(
+            id=event_data.get("id"),
+            node_id=event_data.get("node_id"),
+            url=event_data.get("url"),
+            actor=event_data.get("actor"),
+            event=event_data.get("event"),
+            commit_id=event_data.get("commit_id"),
+            commit_url=event_data.get("commit_url"),
+            created_at=event_data.get("created_at"),
+            performed_via_github_app=event_data.get("performed_via_github_app"),
+        )
+
+        return GitHubReadyForReviewEvent(
+            id=typed_dict["id"],
+            node_id=typed_dict["node_id"],
+            url=typed_dict["url"],
+            actor=(
+                GitHubUser(
+                    **{
+                        k: v
+                        for k, v in typed_dict["actor"].items()
+                        if k in {f.name for f in dataclasses.fields(GitHubUser)}
+                    },
+                )
+                if typed_dict["actor"]
+                else None
+            ),
+            event=typed_dict["event"],
+            commit_id=typed_dict.get("commit_id"),
+            commit_url=typed_dict.get("commit_url"),
+            created_at=typed_dict["created_at"],
+            performed_via_github_app=typed_dict.get("performed_via_github_app"),
+        )
+
+    def _create_review_event(self, event_data: Dict[str, Any]) -> GitHubReviewEvent:
+        """
+        Creates a Review event from raw data.
+
+        Args:
+            event_data: Raw event data
+
+        Returns:
+            GitHubReviewEvent object
+        """
+        typed_dict = GitHubReviewEventFullDict(
+            id=event_data.get("id"),
+            node_id=event_data.get("node_id"),
+            user=event_data.get("user"),
+            body=event_data.get("body"),
+            commit_id=event_data.get("commit_id"),
+            submitted_at=event_data.get("submitted_at"),
+            state=event_data.get("state"),
+            html_url=event_data.get("html_url"),
+            pull_request_url=event_data.get("pull_request_url"),
+            author_association=event_data.get("author_association"),
+            _links=event_data.get("_links"),
+            event=event_data.get("event"),
+        )
+
+        return GitHubReviewEvent(
+            id=typed_dict["id"],
+            node_id=typed_dict["node_id"],
+            user=(
+                GitHubUser(
+                    **{
+                        k: v
+                        for k, v in typed_dict["user"].items()
+                        if k in {f.name for f in dataclasses.fields(GitHubUser)}
+                    },
+                )
+                if typed_dict["user"]
+                else None
+            ),
+            body=typed_dict["body"],
+            commit_id=typed_dict["commit_id"],
+            submitted_at=typed_dict["submitted_at"],
+            state=typed_dict["state"],
+            html_url=typed_dict["html_url"],
+            pull_request_url=typed_dict["pull_request_url"],
+            author_association=typed_dict["author_association"],
+            _links=typed_dict["_links"],
+            event=typed_dict["event"],
+        )
+
     def get_contributors(
         self, org_login: str, repo_name: str
     ) -> List[GitHubContributor]:

backend/analytics_server/mhq/exapi/models/timeline.py

@@ -0,0 +1,115 @@
+from typing import Optional, Dict, Literal, Union
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass
+class GitHubUser:
+    """GitHub user information.
+
+    Required fields:
+        login: GitHub username
+        id: Unique user identifier
+        node_id: GitHub node identifier
+        type: User type (e.g., "User", "Organization")
+    """
+
+    # Required core fields
+    login: str
+    id: int
+    node_id: str
+    type: str
+
+    # Optional URL fields
+    avatar_url: Optional[str] = None
+    html_url: Optional[str] = None
+    url: Optional[str] = None
+    gravatar_id: Optional[str] = None
+
+    # Optional relation URLs
+    followers_url: Optional[str] = None
+    following_url: Optional[str] = None
+    gists_url: Optional[str] = None
+    starred_url: Optional[str] = None
+    subscriptions_url: Optional[str] = None
+    organizations_url: Optional[str] = None
+    repos_url: Optional[str] = None
+    events_url: Optional[str] = None
+    received_events_url: Optional[str] = None
+
+
+@dataclass
+class GitHubReviewEvent:
+    """GitHub pull request review event.
+
+    Required fields:
+        id: Unique event identifier
+        node_id: GitHub node identifier
+        user: User who performed the review
+        event: Event type (always "reviewed")
+    """
+
+    # Required core fields
+    id: int
+    node_id: str
+    user: GitHubUser
+    event: Literal["reviewed"]
+
+    # Optional content fields
+    body: Optional[str] = None
+    state: Optional[str] = None
+
+    # Optional reference fields
+    commit_id: Optional[str] = None
+    html_url: Optional[str] = None
+    pull_request_url: Optional[str] = None
+    author_association: Optional[str] = None
+
+    # Optional metadata
+    submitted_at: Optional[Union[str, datetime]] = None
+    _links: Optional[Dict] = field(default_factory=dict)
+
+    def __post_init__(self):
+        # Convert string timestamps to datetime objects
+        if isinstance(self.submitted_at, str) and self.submitted_at:
+            self.submitted_at = datetime.fromisoformat(
+                self.submitted_at.replace("Z", "+00:00")
+            )
+
+
+@dataclass
+class GitHubReadyForReviewEvent:
+    """GitHub ready for review event for pull requests.
+
+    Required fields:
+        id: Unique event identifier
+        node_id: GitHub node identifier
+        actor: User who marked PR as ready for review
+        event: Event type (always "ready_for_review")
+    """
+
+    # Required core fields
+    id: int
+    node_id: str
+    actor: GitHubUser
+    event: Literal["ready_for_review"]
+
+    # Optional reference fields
+    url: Optional[str] = None
+    commit_id: Optional[str] = None
+    commit_url: Optional[str] = None
+
+    # Optional metadata
+    created_at: Optional[Union[str, datetime]] = None
+    performed_via_github_app: Optional[str] = None
+
+    def __post_init__(self):
+        # Convert string timestamps to datetime objects
+        if isinstance(self.created_at, str) and self.created_at:
+            self.created_at = datetime.fromisoformat(
+                self.created_at.replace("Z", "+00:00")
+            )
+
+
+# Type alias for timeline events
+GitHubTimeline = Union[GitHubReviewEvent, GitHubReadyForReviewEvent]