feat: Add endpoint to fetch GitHub repository contributors (#4013)

api/integrations/github/client.py

@@ -1,19 +1,21 @@
 import logging
-import typing
 from enum import Enum
+from typing import Any
 
 import requests
 from django.conf import settings
 from github import Auth, Github
-from rest_framework import status
-from rest_framework.response import Response
 
 from integrations.github.constants import (
     GITHUB_API_CALLS_TIMEOUT,
     GITHUB_API_URL,
     GITHUB_API_VERSION,
 )
-from integrations.github.dataclasses import RepoQueryParams
+from integrations.github.dataclasses import (
+    IssueQueryParams,
+    PaginatedQueryParams,
+    RepoQueryParams,
+)
 from integrations.github.models import GithubConfiguration
 
 logger = logging.getLogger(__name__)
@@ -66,9 +68,34 @@ def generate_jwt_token(app_id: int) -> str:  # pragma: no cover
     return token
 
 
+def build_paginated_response(
+    results: list[dict[str, Any]],
+    response: requests.Response,
+    total_count: int | None = None,
+    incomplete_results: bool | None = None,
+) -> dict[str, Any]:
+    data: dict[str, Any] = {
+        "results": results,
+    }
+
+    if response.links.get("prev"):
+        data["previous"] = response.links.get("prev")
+
+    if response.links.get("next"):
+        data["next"] = response.links.get("next")
+
+    if total_count:
+        data["total_count"] = total_count
+
+    if incomplete_results:
+        data["incomplete_results"] = incomplete_results
+
+    return data
+
+
 def post_comment_to_github(
     installation_id: str, owner: str, repo: str, issue: str, body: str
-) -> dict[str, typing.Any]:
+) -> dict[str, Any]:
 
     url = f"{GITHUB_API_URL}repos/{owner}/{repo}/issues/{issue}/comments"
     headers = build_request_headers(installation_id)
@@ -89,11 +116,11 @@ def delete_github_installation(installation_id: str) -> requests.Response:
     return response
 
 
-def fetch_github_resource(
+def fetch_search_github_resource(
     resource_type: ResourceType,
     organisation_id: int,
-    params: RepoQueryParams,
-) -> dict[str, typing.Any]:
+    params: IssueQueryParams,
+) -> dict[str, Any]:
     github_configuration = GithubConfiguration.objects.get(
         organisation_id=organisation_id, deleted_at__isnull=True
     )
@@ -135,22 +162,23 @@ def fetch_github_resource(
         }
         for i in json_response["items"]
     ]
-    data = {
-        "results": results,
-        "count": json_response["total_count"],
-        "incomplete_results": json_response["incomplete_results"],
-    }
-    if response.links.get("prev"):
-        data["previous"] = response.links.get("prev")
 
-    if response.links.get("next"):
-        data["next"] = response.links.get("next")
-
-    return data
+    return build_paginated_response(
+        results=results,
+        response=response,
+        total_count=json_response["total_count"],
+        incomplete_results=json_response["incomplete_results"],
+    )
 
 
-def fetch_github_repositories(installation_id: str) -> Response:
-    url = f"{GITHUB_API_URL}installation/repositories"
+def fetch_github_repositories(
+    installation_id: str,
+    params: PaginatedQueryParams,
+) -> dict[str, Any]:
+    url = (
+        f"{GITHUB_API_URL}installation/repositories?"
+        + f"&per_page={params.page_size}&page={params.page}"
+    )
 
     headers: dict[str, str] = build_request_headers(installation_id)
 
@@ -165,15 +193,8 @@ def fetch_github_repositories(installation_id: str) -> Response:
         }
         for i in json_response["repositories"]
     ]
-    data = {
-        "repositories": results,
-        "total_count": json_response["total_count"],
-    }
-    return Response(
-        data=data,
-        content_type="application/json",
-        status=status.HTTP_200_OK,
-    )
+
+    return build_paginated_response(results, response, json_response["total_count"])
 
 
 def get_github_issue_pr_title_and_state(
@@ -191,5 +212,35 @@ def get_github_issue_pr_title_and_state(
     headers = build_request_headers(installation_id)
     response = requests.get(url, headers=headers, timeout=GITHUB_API_CALLS_TIMEOUT)
     response.raise_for_status()
-    response_json = response.json()
-    return {"title": response_json["title"], "state": response_json["state"]}
+    json_response = response.json()
+    return {"title": json_response["title"], "state": json_response["state"]}
+
+
+def fetch_github_repo_contributors(
+    organisation_id: int,
+    params: RepoQueryParams,
+) -> dict[str, Any]:
+    installation_id = GithubConfiguration.objects.get(
+        organisation_id=organisation_id, deleted_at__isnull=True
+    ).installation_id
+
+    url = (
+        f"{GITHUB_API_URL}repos/{params.repo_owner}/{params.repo_name}/contributors?"
+        + f"&per_page={params.page_size}&page={params.page}"
+    )
+
+    headers = build_request_headers(installation_id)
+    response = requests.get(url, headers=headers, timeout=GITHUB_API_CALLS_TIMEOUT)
+    response.raise_for_status()
+    json_response = response.json()
+
+    results = [
+        {
+            "login": i["login"],
+            "avatar_url": i["avatar_url"],
+            "contributions": i["contributions"],
+        }
+        for i in json_response
+    ]
+
+    return build_paginated_response(results, response)

api/integrations/github/dataclasses.py

@@ -1,44 +1,55 @@
-import typing
-from dataclasses import dataclass
-from typing import Optional
+from dataclasses import dataclass, field
+from typing import Any, Optional
 
 
+# Base Dataclasses
 @dataclass
 class GithubData:
     installation_id: str
     feature_id: int
     feature_name: str
     type: str
-    feature_states: list[dict[str, typing.Any]] | None = None
+    feature_states: list[dict[str, Any]] | None = None
     url: str | None = None
     project_id: int | None = None
     segment_name: str | None = None
 
     @classmethod
-    def from_dict(cls, data_dict: dict[str, typing.Any]) -> "GithubData":
+    def from_dict(cls, data_dict: dict[str, Any]) -> "GithubData":
         return cls(**data_dict)
 
 
 @dataclass
 class CallGithubData:
     event_type: str
     github_data: GithubData
-    feature_external_resources: list[dict[str, typing.Any]]
+    feature_external_resources: list[dict[str, Any]]
 
 
+# Dataclasses for external calls to GitHub API
 @dataclass
-class RepoQueryParams:
+class PaginatedQueryParams:
+    page: int = field(default=1, kw_only=True)
+    page_size: int = field(default=100, kw_only=True)
+
+    def __post_init__(self):
+        if self.page < 1:
+            raise ValueError("Page must be greater or equal than 1")
+        if self.page_size < 1 or self.page_size > 100:
+            raise ValueError("Page size must be an integer between 1 and 100")
+
+
+@dataclass
+class RepoQueryParams(PaginatedQueryParams):
     repo_owner: str
     repo_name: str
+
+
+@dataclass
+class IssueQueryParams(RepoQueryParams):
     search_text: Optional[str] = None
-    page: Optional[int] = 1
-    page_size: Optional[int] = 100
     state: Optional[str] = "open"
     author: Optional[str] = None
     assignee: Optional[str] = None
     search_in_body: Optional[bool] = True
     search_in_comments: Optional[bool] = False
-
-    @classmethod
-    def from_dict(cls, data_dict: dict[str, typing.Any]) -> "RepoQueryParams":
-        return cls(**data_dict)

api/integrations/github/serializers.py

@@ -2,7 +2,11 @@
 from rest_framework.serializers import ModelSerializer
 from rest_framework_dataclasses.serializers import DataclassSerializer
 
-from integrations.github.dataclasses import RepoQueryParams
+from integrations.github.dataclasses import (
+    IssueQueryParams,
+    PaginatedQueryParams,
+    RepoQueryParams,
+)
 from integrations.github.models import GithubConfiguration, GithubRepository
 
 
@@ -30,8 +34,18 @@ class Meta:
         )
 
 
+class PaginatedQueryParamsSerializer(DataclassSerializer):
+    class Meta:
+        dataclass = PaginatedQueryParams
+
+
 class RepoQueryParamsSerializer(DataclassSerializer):
     class Meta:
         dataclass = RepoQueryParams
 
+
+class IssueQueryParamsSerializer(DataclassSerializer):
+    class Meta:
+        dataclass = IssueQueryParams
+
     search_in_body = serializers.BooleanField(required=False, default=True)