from __future__ import annotations import logging import random import threading import time from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from github import Auth, Github, GithubException, RateLimitExceededException logger = logging.getLogger(__name__) _RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504} _DEFAULT_MAX_RETRIES = 3 _DEFAULT_BACKOFF_BASE = 2.0 _DEFAULT_BACKOFF_MAX = 60.0 class GitHubGraphQLError(RuntimeError): """Raised when GitHub GraphQL returns an `errors` payload.""" def __init__(self, errors: list[Any]): self.errors = errors super().__init__(f"GitHub GraphQL error: {errors}") @dataclass class AuthConfig: """Encapsulates either PAT or GitHub App credentials.""" token: str | None = None app_id: int | None = None private_key: str | None = None installation_id: int | None = None _github: Github | None = field(default=None, init=False, repr=False) _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) def build_github(self) -> Github: if self._github is not None: return self._github with self._lock: if self._github is not None: return self._github if self.app_id and self.private_key and self.installation_id: app_auth = Auth.AppAuth( app_id=self.app_id, private_key=self.private_key, ) gh = Github( auth=Auth.AppInstallationAuth( app_auth=app_auth, installation_id=self.installation_id ) ) elif self.token: gh = Github(auth=Auth.Token(self.token)) else: raise ValueError("Provide either token or app_id + private_key + installation_id") self._github = gh return gh @dataclass class GitHubBaseClient: auth_config: AuthConfig def __post_init__(self) -> None: self._gh = self.auth_config.build_github() def _graphql_once( self, query: str, variables: dict[str, Any] | None = None, ) -> dict[str, Any]: headers, data = self._gh.requester.requestJsonAndCheck( "POST", "/graphql", input={ "query": query, "variables": variables or {}, }, ) logger.debug( "GitHub GraphQL rate limit: remaining=%s used=%s limit=%s reset=%s resource=%s request-id=%s", headers.get("x-ratelimit-remaining"), headers.get("x-ratelimit-used"), headers.get("x-ratelimit-limit"), headers.get("x-ratelimit-reset"), headers.get("x-ratelimit-resource"), headers.get("x-github-request-id"), ) if not data: raise RuntimeError("Empty response from GitHub GraphQL") errors = data.get("errors") if errors: logger.error("GitHub GraphQL returned errors: %s", errors) raise GitHubGraphQLError(errors) payload = data.get("data") if not isinstance(payload, dict): raise RuntimeError("Invalid GraphQL response format") return payload def _graphql_with_retry( self, query: str, variables: dict[str, Any] | None = None, *, max_retries: int = _DEFAULT_MAX_RETRIES, backoff_base: float = _DEFAULT_BACKOFF_BASE, backoff_max: float = _DEFAULT_BACKOFF_MAX, ) -> dict[str, Any]: last_exc: Exception | None = None for attempt in range(max_retries + 1): is_last_attempt = attempt == max_retries try: return self._graphql_once(query, variables) except RateLimitExceededException as exc: last_exc = exc if is_last_attempt: break wait = self._compute_rate_limit_wait( reset_at=getattr(exc, "reset_time", None), attempt=attempt, backoff_base=backoff_base, backoff_max=backoff_max, ) logger.warning( "GitHub primary rate limit hit (attempt %d/%d) — waiting %.1fs before retry", attempt + 1, max_retries + 1, wait, ) time.sleep(wait) except GithubException as exc: if not self._is_retryable_github_exception(exc): raise last_exc = exc if is_last_attempt: break wait = self._compute_github_exception_wait( exc=exc, attempt=attempt, backoff_base=backoff_base, backoff_max=backoff_max, ) logger.warning( "GitHub API error status=%s (attempt %d/%d) — waiting %.1fs before retry", getattr(exc, "status", None), attempt + 1, max_retries + 1, wait, ) time.sleep(wait) except GitHubGraphQLError as exc: if not self._is_retryable_graphql_error(exc): raise last_exc = exc if is_last_attempt: break wait = min( backoff_base * (2**attempt) + random.uniform(0, 1), backoff_max, ) logger.warning( "GitHub GraphQL returned retryable errors (attempt %d/%d) — waiting %.1fs before retry", attempt + 1, max_retries + 1, wait, ) time.sleep(wait) logger.error("GitHub GraphQL request failed after %d attempts", max_retries + 1) if last_exc is not None: raise last_exc raise RuntimeError("GitHub GraphQL request failed without a captured exception") def _is_retryable_github_exception(self, exc: GithubException) -> bool: status = getattr(exc, "status", None) if status in _RETRYABLE_STATUS_CODES: return True message = str(exc).lower() return ( "secondary rate limit" in message or "abuse detection" in message or "temporarily unavailable" in message ) def _is_retryable_graphql_error(self, exc: GitHubGraphQLError) -> bool: for error in exc.errors: if not isinstance(error, dict): continue message = str(error.get("message", "")).lower() error_type = str(error.get("type", "")).lower() if ( "rate limit" in message or "secondary rate limit" in message or "something went wrong" in message or error_type in {"rate_limit", "service_unavailable"} ): return True return False def _compute_github_exception_wait( self, *, exc: GithubException, attempt: int, backoff_base: float, backoff_max: float, ) -> float: retry_after = self._parse_retry_after(exc) if retry_after is not None: wait = float(retry_after) + 1.0 return float(min(wait, backoff_max)) wait = backoff_base * (2**attempt) + random.uniform(0, 1) return float(min(wait, backoff_max)) def _compute_rate_limit_wait( self, *, reset_at: Any, attempt: int, backoff_base: float, backoff_max: float, ) -> float: if isinstance(reset_at, datetime): if reset_at.tzinfo is None: reset_at = reset_at.replace(tzinfo=UTC) wait = max(0.0, reset_at.timestamp() - time.time()) + 1.0 return float(min(wait, backoff_max)) if isinstance(reset_at, (int, float)): wait = max(0.0, float(reset_at) - time.time()) + 1.0 return float(min(wait, backoff_max)) wait = backoff_base * (2**attempt) return float(min(wait, backoff_max)) def _parse_retry_after(self, exc: GithubException) -> int | None: try: headers = getattr(exc, "headers", None) or {} value = headers.get("retry-after") or headers.get("Retry-After") return int(value) if value is not None else None except (ValueError, TypeError, AttributeError): return None