github_client.py

import requests
import time
from typing import List, Dict, Optional
from config import REPO_OWNER, REPO_NAME, GITHUB_TOKEN


class GitHubClient:
    def __init__(self, token: str = GITHUB_TOKEN):
        self.token = GITHUB_TOKEN
        self.api_url = "https://api.github.com"
        self.headers = {
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "vllm-issue-monitor",
            "X-GitHub-Api-Version": "2022-11-28"
        }
        if token:
            self.headers["Authorization"] = f"token {token}"

    def _make_request(self, url: str, params: Optional[Dict] = None, max_retries: int = 3) -> requests.Response:
        for attempt in range(max_retries):
            try:
                response = requests.get(url, headers=self.headers, params=params, timeout=30, allow_redirects=True)
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
        raise requests.exceptions.RequestException("Max retries exceeded")

    def search_issues(self, query: str, per_page: int = 100) -> List[Dict]:
        issues = []
        page = 1
        
        while True:
            url = f"{self.api_url}/search/issues"
            params = {
                "q": query,
                "per_page": per_page,
                "page": page
            }
            
            response = self._make_request(url, params)
            data = response.json()
            items = data.get("items", [])
            
            if not items:
                break
                
            issues.extend(items)
            page += 1
            
            if len(items) < per_page:
                break
                
        return issues

    def get_issues(self, state: str = "open", per_page: int = 100, 
                  labels: Optional[str] = None, since: Optional[str] = None) -> List[Dict]:
        query = f"repo:{REPO_OWNER}/{REPO_NAME} is:issue is:{state}"
        
        if labels:
            query += f" labels:{labels}"
        
        if since:
            query += f" since:{since}"
        
        return self.search_issues(query, per_page)

    def get_repository_issues(self, state: str = "open", per_page: int = 100,
                           labels: Optional[str] = None, since: Optional[str] = None) -> List[Dict]:
        issues = []
        page = 1
        
        url = f"{self.api_url}/repos/{REPO_OWNER}/{REPO_NAME}/issues"
        params = {
            "state": state,
            "per_page": per_page,
            "page": page
        }
        
        if labels:
            params["labels"] = labels
        
        if since:
            params["since"] = since
        
        while True:
            response = self._make_request(url, params)
            data = response.json()
            
            if not data:
                break
                
            issues.extend(data)
            page += 1
            
            if len(data) < per_page:
                break
                
            params["page"] = page
                
        return issues

    def get_issue_comments(self, issue_number: int) -> List[Dict]:
        url = f"{self.api_url}/repos/{REPO_OWNER}/{REPO_NAME}/issues/{issue_number}/comments"
        try:
            response = self._make_request(url)
            return response.json()
        except Exception:
            return []