""" Gitea API client for Kanbante. """ import os from typing import Any import httpx from dotenv import load_dotenv load_dotenv(os.path.expanduser("~/workspace/.env")) class GiteaClient: """Lightweight Gitea API client for Kanbante.""" def __init__(self) -> None: self.base_url = os.getenv("GITEA_URL", "https://git.dracodev.net") self.token = os.getenv("GITEA_TOKEN", "") self.client = httpx.Client( base_url=f"{self.base_url}/api/v1", headers={ "Authorization": f"token {self.token}", "Accept": "application/json", }, timeout=15.0, ) def _get(self, path: str, **params: Any) -> dict | list: r = self.client.get(path, params=params) r.raise_for_status() return r.json() def get_user_repos(self, username: str = "bruno") -> list[dict]: """Get all repos for a user.""" return self._get(f"/users/{username}/repos", limit=100) def search_repos(self, uid: int = 1) -> list[dict]: """Search all repos visible to user.""" result = self._get("/repos/search", uid=uid, limit=200) if isinstance(result, dict) and "data" in result: return result["data"] return result if isinstance(result, list) else [] def get_org_repos(self, org: str) -> list[dict]: """Get repos for an organization.""" return self._get(f"/orgs/{org}/repos", limit=50) def get_repo_issues( self, owner: str, repo: str, state: str = "all" ) -> list[dict]: """Get issues for a repo.""" return self._get( f"/repos/{owner}/{repo}/issues", state=state, limit=100, ) def get_all_issues( self, repos: list[dict], state: str = "open" ) -> dict[str, list[dict]]: """Get issues for multiple repos, keyed by full_name.""" issues_by_repo: dict[str, list[dict]] = {} for repo in repos: full_name = repo["full_name"] owner, name = full_name.split("/", 1) try: issues = self.get_repo_issues(owner, name, state=state) issues_by_repo[full_name] = issues except Exception: issues_by_repo[full_name] = [] return issues_by_repo def get_user(self) -> dict: """Get authenticated user info.""" return self._get("/user")