Bring recent diffs for GitHub connector into develop branch (#3780)

PR to bring the recent changes from
#3708 into the `develop`
branch.

This PR is a prerequisite to splitting apart the GitHub connector

Author: mattnowzari

app/connectors_service/connectors/sources/github.py

@@ -7,6 +7,7 @@
 
 import json
 import time
+from collections import defaultdict
 from enum import Enum
 from functools import cached_property, partial
 
@@ -19,9 +20,7 @@
 )
 from connectors_sdk.logger import logger
 from connectors_sdk.source import BaseDataSource, ConfigurableFieldValueError
-from connectors_sdk.utils import (
-    nested_get_from_dict,
-)
+from connectors_sdk.utils import nested_get_from_dict
 from gidgethub import QueryError, sansio
 from gidgethub.abc import (
     BadGraphQLRequest,
@@ -199,6 +198,11 @@ class GithubQuery(Enum):
     }
     }
     """
+    BATCH_REPO_QUERY_TEMPLATE = """
+    query ({batch_queries}) {{
+        {query_body}
+    }}
+    """
     PULL_REQUEST_QUERY = f"""
     query ($owner: String!, $name: String!, $cursor: String) {{
     repository(owner: $owner, name: $name) {{
@@ -770,12 +774,18 @@ def _get_client(self):
         strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
         skipped_exceptions=UnauthorizedException,
     )
-    async def graphql(self, query, variables=None):
+    async def graphql(
+        self,
+        query: str,
+        variables: dict | None = None,
+        ignore_errors: list[str] | None = None,
+    ) -> dict:
         """Invoke GraphQL request to fetch repositories, pull requests, and issues.
 
         Args:
             query: Dictionary comprising of query for the GraphQL request.
             variables: Dictionary comprising of query for the GraphQL request.
+            ignore_errors: List of error types to ignore and return null for instead of raising exceptions.
 
         Raises:
             UnauthorizedException: Unauthorized exception
@@ -784,6 +794,9 @@ async def graphql(self, query, variables=None):
         Yields:
             dictionary: Client response
         """
+
+        ignore_errors = ignore_errors or []
+
         url = f"{self.base_url}/graphql"
         self._logger.debug(
             f"Sending POST to {url} with query: '{json.dumps(query)}' and variables: '{json.dumps(variables)}'"
@@ -809,13 +822,24 @@ async def graphql(self, query, variables=None):
             else:
                 raise
         except QueryError as exception:
-            for error in exception.response.get("errors"):
+            errors = exception.response.get("errors", [])
+            for error in errors:
                 if (
-                    error.get("type").lower() == "rate_limited"
+                    error.get("type", "").lower() == "rate_limited"
                     and "api rate limit exceeded" in error.get("message").lower()
                 ):
                     await self._put_to_sleep(resource_type="graphql")
-            msg = f"Error while executing query. Exception: {exception.response.get('errors')}"
+
+            if all(error.get("type") in ignore_errors for error in errors):
+                # All errors are ignored, return just the data part without errors
+                return exception.response.get("data", {})
+
+            # report only non-ignored errors
+            non_ignored_errors = [
+                error for error in errors if error.get("type") not in ignore_errors
+            ]
+
+            msg = f"Error while executing query. Exception: {non_ignored_errors}"
             raise Exception(msg) from exception
         except Exception as e:
             self._logger.debug(
@@ -1004,6 +1028,93 @@ async def get_foreign_repo(self, repo_name):
         )
         return data.get(REPOSITORY_OBJECT)
 
+    async def get_repos_by_fully_qualified_name_batch(
+        self, repo_names: list[str], batch_size: int = 15
+    ) -> dict[str, dict | None]:
+        """Batch validate multiple repositories by fully qualified name in fewer GraphQL requests.
+
+        Args:
+            repo_names (list): List of fully qualified repository names (owner/repo format) to validate
+            batch_size (int): Number of repos to validate per request (default 15)
+
+        Returns:
+            dict: Dictionary mapping repo_name to repository object (or None if invalid)
+        """
+        results = {}
+
+        # Process repositories in batches to avoid hitting GraphQL complexity limits
+        for i in range(0, len(repo_names), batch_size):
+            batch = repo_names[i : i + batch_size]
+            batch_results = await self._fetch_repos_batch(batch)
+            results.update(batch_results)
+
+        return results
+
+    async def _fetch_repos_batch(self, repo_names: list[str]) -> dict[str, dict | None]:
+        """Fetch a batch of repositories in a single GraphQL query using aliases.
+
+        Note: This method will NOT raise exceptions for non-existent repositories.
+        Invalid/non-existent repositories will be returned as None values in the result.
+        """
+        if not repo_names:
+            return {}
+
+        query_parts = []
+        variables = {}
+
+        for i, repo_name in enumerate(repo_names):
+            owner, repo = self.get_repo_details(repo_name=repo_name)
+            alias = f"repo{i}"
+
+            variables[f"{alias}_owner"] = owner
+            variables[f"{alias}_name"] = repo
+
+            query_part = f"""
+            {alias}: repository(owner: ${alias}_owner, name: ${alias}_name) {{
+                id
+                updatedAt
+                name
+                nameWithOwner
+                url
+                description
+                visibility
+                primaryLanguage {{
+                    name
+                }}
+                defaultBranchRef {{
+                    name
+                }}
+                isFork
+                stargazerCount
+                watchers {{
+                    totalCount
+                }}
+                forkCount
+                createdAt
+                isArchived
+            }}"""
+            query_parts.append(query_part)
+
+        variable_declarations = [
+            f"${var_name}: String!" for var_name in variables.keys()
+        ]
+
+        query = GithubQuery.BATCH_REPO_QUERY_TEMPLATE.value.format(
+            batch_queries=", ".join(variable_declarations),
+            query_body=" ".join(query_parts),
+        )
+
+        data = await self.graphql(
+            query=query, variables=variables, ignore_errors=["NOT_FOUND"]
+        )
+
+        # Map results back to repo names
+        results = {
+            repo_name: data.get(f"repo{i}") for i, repo_name in enumerate(repo_names)
+        }
+
+        return results
+
     async def _fetch_all_members(self, org_name):
         org_variables = {
             "orgName": org_name,
@@ -1318,7 +1429,7 @@ async def get_invalid_repos(self):
         else:
             return await self._get_invalid_repos_for_personal_access_token()
 
-    async def _get_invalid_repos_for_github_app(self):
+    async def _get_invalid_repos_for_github_app(self) -> list[str]:
         # A github app can be installed on multiple orgs/personal accounts,
         # so the repo must be configured in the format of 'OWNER/REPO', any other format will be rejected
         invalid_repos = set(
@@ -1329,18 +1440,62 @@ async def _get_invalid_repos_for_github_app(self):
                 self.configured_repos,
             )
         )
-        for full_repo_name in self.configured_repos:
-            if full_repo_name in invalid_repos:
-                continue
+
+        # Group valid repos by owner for batch validation
+        valid_repos = [
+            repo for repo in self.configured_repos if repo not in invalid_repos
+        ]
+        repos_by_owner = defaultdict(list)
+
+        await self._fetch_installations()
+
+        for full_repo_name in valid_repos:
             owner, repo_name = self.github_client.get_repo_details(
                 repo_name=full_repo_name
             )
-            if await self._get_repo_object_for_github_app(owner, repo_name) is None:
+
+            # Check if GitHub App is installed on this owner
+            if owner not in self._installations:
+                self._logger.debug(
+                    f"Invalid repo {full_repo_name} as the github app is not installed on {owner}"
+                )
                 invalid_repos.add(full_repo_name)
+                continue
+
+            repos_by_owner[owner].append(full_repo_name)
+
+        # Batch validate repos for each owner
+        for owner, owner_repos in repos_by_owner.items():
+            await self.github_client.update_installation_id(self._installations[owner])
+
+            batch_results = (
+                await self.github_client.get_repos_by_fully_qualified_name_batch(
+                    owner_repos
+                )
+            )
+
+            for repo_name, repo_data in batch_results.items():
+                if repo_data:
+                    if self.configuration["repo_type"] == "organization":
+                        if owner not in self.org_repos:
+                            self.org_repos[owner] = {}
+                        self.org_repos[owner][repo_name] = repo_data
+                    else:
+                        if owner not in self.user_repos:
+                            self.user_repos[owner] = {}
+                        self.user_repos[owner][repo_name] = repo_data
+                else:
+                    self._logger.debug(f"Detected invalid repository: {repo_name}.")
+                    invalid_repos.add(repo_name)
 
         return list(invalid_repos)
 
-    async def _get_repo_object_for_github_app(self, owner, repo_name):
+    async def _get_repo_object_for_github_app(
+        self, owner: str, repo_name: str
+    ) -> dict | None:
+        # Note: this method fetches potentially all user or org repos and caches them,
+        # so it should be used sparingly as it could consume a lot of API rate limit
+        # due to possibly multiple pages of repos
         await self._fetch_installations()
         full_repo_name = f"{owner}/{repo_name}"
         if owner not in self._installations:
@@ -1373,50 +1528,48 @@ async def _get_repo_object_for_github_app(self, owner, repo_name):
 
         return cached_repo[owner][full_repo_name]
 
-    async def _get_invalid_repos_for_personal_access_token(self):
+    async def _get_invalid_repos_for_personal_access_token(self) -> list[str]:
         try:
+            all_repos: list[str] = []
+            # Combine all repos for unified batch validation
             if self.configuration["repo_type"] == "other":
                 logged_in_user = await self._logged_in_user()
                 foreign_repos, configured_repos = self.github_client.bifurcate_repos(
                     repos=self.configured_repos,
                     owner=logged_in_user,
                 )
-                if logged_in_user not in self.user_repos:
-                    self.user_repos[logged_in_user] = {}
-                async for repo in self.github_client.get_user_repos(logged_in_user):
-                    self.user_repos[logged_in_user][repo["nameWithOwner"]] = repo
-                invalid_repos = list(
-                    set(configured_repos) - set(self.user_repos[logged_in_user].keys())
-                )
-
-                for repo_name in foreign_repos:
-                    try:
-                        self.foreign_repos[
-                            repo_name
-                        ] = await self.github_client.get_foreign_repo(
-                            repo_name=repo_name
-                        )
-                    except Exception:
-                        self._logger.debug(f"Detected invalid repository: {repo_name}.")
-                        invalid_repos.append(repo_name)
+                all_repos = configured_repos + foreign_repos
             else:
                 foreign_repos, configured_repos = self.github_client.bifurcate_repos(
                     repos=self.configured_repos,
                     owner=self.configuration["org_name"],
                 )
-                configured_repos.extend(foreign_repos)
-                if self.configuration["org_name"] not in self.org_repos:
-                    self.org_repos[self.configuration["org_name"]] = {}
-                async for repo in self.github_client.get_org_repos(
-                    self.configuration["org_name"]
-                ):
-                    self.org_repos[self.configuration["org_name"]][
-                        repo["nameWithOwner"]
-                    ] = repo
-                invalid_repos = list(
-                    set(configured_repos)
-                    - set(self.org_repos[self.configuration["org_name"]].keys())
+                all_repos = configured_repos + foreign_repos
+
+            invalid_repos = []
+            batch_results = (
+                await self.github_client.get_repos_by_fully_qualified_name_batch(
+                    all_repos
                 )
+            )
+            for repo_name, repo_data in batch_results.items():
+                if not repo_data:
+                    self._logger.debug(f"Detected invalid repository: {repo_name}.")
+                    invalid_repos.append(repo_name)
+                    continue
+                # Store valid repos for potential later use
+                if repo_name in foreign_repos:
+                    self.foreign_repos[repo_name] = repo_data
+                else:
+                    # Store configured repos in the appropriate cache
+                    if self.configuration["repo_type"] == "other":
+                        logged_in_user = await self._logged_in_user()
+                        self.user_repos.setdefault(logged_in_user, {})[repo_name] = (
+                            repo_data
+                        )
+                    else:
+                        org_name = self.configuration["org_name"]
+                        self.org_repos.setdefault(org_name, {})[repo_name] = repo_data
             return invalid_repos
         except Exception as exception:
             self._logger.exception(
@@ -1588,6 +1741,9 @@ async def _fetch_installations(self):
         return self._installations
 
     async def _get_personal_repos(self, user):
+        # Note: this method fetches potentially all user repos and caches them,
+        # so it should be used sparingly as it could consume a lot of API rate limit
+        # due to possibly multiple pages of repos
         self._logger.info(f"Fetching personal repos {user}")
         if user in self.user_repos:
             for repo_object in self.user_repos[user]:
@@ -1599,6 +1755,9 @@ async def _get_personal_repos(self, user):
                 yield repo_object
 
     async def _get_org_repos(self, org_name):
+        # Note: this method fetches potentially all org repos and caches them,
+        # so it should be used sparingly as it could consume a lot of API rate limit
+        # due to possibly multiple pages of repos
         self._logger.info(f"Fetching org repos for {org_name}")
         if org_name in self.org_repos:
             for repo_object in self.org_repos[org_name]:
@@ -1610,6 +1769,8 @@ async def _get_org_repos(self, org_name):
                 yield repo_object
 
     async def _get_configured_repos(self, configured_repos):
+        # Note: this method calls _get_repo_object_for_github_app - see comments there
+        # for potential performance implications
         self._logger.info(f"Fetching configured repos: '{configured_repos}'")
         for repo_name in configured_repos:
             self._logger.info(f"Fetching repo: '{repo_name}'")
@@ -1649,6 +1810,10 @@ async def _get_configured_repos(self, configured_repos):
             yield self._convert_repo_object_to_doc(repo_object)
 
     async def _fetch_repos(self):
+        # Note: this method calls _get_configured_repos
+        # which in turn calls _get_repo_object_for_github_app
+        # or _get_personal_repos/_get_org_repos, see comments there
+        # for potential performance/rate limit implications
         self._logger.info("Fetching repos")
         try:
             if (