Initial upload: DevDash CLI with TUI dashboard
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
This commit is contained in:
38
src/api/github.py
Normal file
38
src/api/github.py
Normal file
@@ -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]
|
||||||
Reference in New Issue
Block a user