|
2 | 2 |
|
3 | 3 | import logging |
4 | 4 | import re |
5 | | -from collections.abc import Callable, Mapping, MutableMapping, Sequence |
| 5 | +from collections.abc import Callable, Iterable, Mapping, MutableMapping, Sequence |
6 | 6 | from dataclasses import dataclass |
7 | 7 | from enum import StrEnum |
8 | 8 | from typing import Any, NotRequired, TypedDict |
@@ -325,46 +325,55 @@ def get_repositories( |
325 | 325 | query: str | None = None, |
326 | 326 | page_number_limit: int | None = None, |
327 | 327 | accessible_only: bool = False, |
| 328 | + use_cache: bool = False, |
328 | 329 | ) -> list[RepositoryInfo]: |
329 | 330 | """ |
330 | 331 | args: |
331 | 332 | * query - a query to filter the repositories by |
332 | 333 | * accessible_only - when True with a query, fetch only installation- |
333 | 334 | accessible repos and filter locally instead of using the Search API |
334 | 335 | (which may return repos outside the installation's scope) |
| 336 | + * use_cache - when True, serve repos from a short-lived cache instead |
| 337 | + of re-fetching all pages from GitHub on every call |
335 | 338 |
|
336 | 339 | This fetches all repositories accessible to the Github App |
337 | 340 | https://docs.github.com/en/rest/apps/installations#list-repositories-accessible-to-the-app-installation |
338 | 341 | """ |
339 | | - if not query or accessible_only: |
340 | | - all_repos = self.get_client().get_repos(page_number_limit=page_number_limit) |
341 | | - repos: list[RepositoryInfo] = [ |
| 342 | + client = self.get_client() |
| 343 | + |
| 344 | + def to_repo_info(raw_repos: Iterable[Mapping[str, Any]]) -> list[RepositoryInfo]: |
| 345 | + return [ |
342 | 346 | { |
343 | 347 | "name": i["name"], |
344 | 348 | "identifier": i["full_name"], |
345 | 349 | "external_id": self.get_repo_external_id(i), |
346 | 350 | "default_branch": i.get("default_branch"), |
347 | 351 | } |
348 | | - for i in all_repos |
349 | | - if not i.get("archived") |
| 352 | + for i in raw_repos |
350 | 353 | ] |
351 | | - if query: |
352 | | - query_lower = query.lower() |
353 | | - repos = [r for r in repos if query_lower in str(r["identifier"]).lower()] |
354 | | - return repos |
355 | 354 |
|
| 355 | + def _get_all_repos(): |
| 356 | + if use_cache: |
| 357 | + return client.get_repos_cached() |
| 358 | + return client.get_repos(page_number_limit=page_number_limit) |
| 359 | + |
| 360 | + if not query: |
| 361 | + all_repos = _get_all_repos() |
| 362 | + return to_repo_info(r for r in all_repos if not r.get("archived")) |
| 363 | + |
| 364 | + if accessible_only: |
| 365 | + all_repos = _get_all_repos() |
| 366 | + query_lower = query.lower() |
| 367 | + return to_repo_info( |
| 368 | + r |
| 369 | + for r in all_repos |
| 370 | + if not r.get("archived") and query_lower in r["full_name"].lower() |
| 371 | + ) |
| 372 | + |
| 373 | + assert not use_cache, "use_cache is not supported with the Search API path" |
356 | 374 | full_query = build_repository_query(self.model.metadata, self.model.name, query) |
357 | | - response = self.get_client().search_repositories(full_query) |
358 | | - search_repos: list[RepositoryInfo] = [ |
359 | | - { |
360 | | - "name": i["name"], |
361 | | - "identifier": i["full_name"], |
362 | | - "external_id": self.get_repo_external_id(i), |
363 | | - "default_branch": i.get("default_branch"), |
364 | | - } |
365 | | - for i in response.get("items", []) |
366 | | - ] |
367 | | - return search_repos |
| 375 | + response = client.search_repositories(full_query) |
| 376 | + return to_repo_info(response.get("items", [])) |
368 | 377 |
|
369 | 378 | def get_unmigratable_repositories(self) -> list[RpcRepository]: |
370 | 379 | accessible_repos = self.get_repositories() |
|
0 commit comments