diff --git a/src/api/github.py b/src/api/github.py new file mode 100644 index 0000000..64e568a --- /dev/null +++ b/src/api/github.py @@ -0,0 +1,38 @@ +from typing import Any, Dict, List +import httpx +from src.api.base import APIClient + + +class GitHubClient(APIClient): + def __init__(self, token: str = None): + super().__init__("https://api.github.com", token or "") + + async def get_pull_requests(self, owner: str, repo: str) -> List[Dict[str, Any]]: + url = f"{self.base_url}/repos/{owner}/{repo}/pulls?state=open&per_page=10" + headers = self.headers.copy() + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=headers) + response.raise_for_status() + return response.json() + + async def get_issues(self, owner: str, repo: str) -> List[Dict[str, Any]]: + url = f"{self.base_url}/repos/{owner}/{repo}/issues?state=open&per_page=10" + headers = self.headers.copy() + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=headers) + response.raise_for_status() + return [i for i in response.json() if "pull_request" not in i] + + async def get_workflows(self, owner: str, repo: str) -> List[Dict[str, Any]]: + url = f"{self.base_url}/repos/{owner}/{repo}/actions/workflows" + headers = self.headers.copy() + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=headers) + response.raise_for_status() + return response.json().get("workflows", [])[:5]