kanbante/app/github_client.py

"""GitHub forge client (0.24.0, P1).

Subclasses GiteaClient and overrides only what differs on
GitHub's REST API:

- base URL without suffix (api.github.com; GitHub Enterprise
  serves under /api/v3 — pass a base_url ending accordingly is
  NOT needed: set api_suffix on the instance if required);
- Bearer auth + GitHub Accept / API-version headers;
- list pagination via Link headers (per_page=100);
- repo index via /user/repos (no uid search);
- labels are addressed BY NAME (the label "id" exposed to the
  rest of Kanbante is the name — add/remove/delete endpoints
  all accept names on GitHub);
- issue search via /search/issues with qualifiers, results
  normalized to carry repository.full_name like Gitea's;
- archiving via PATCH (Gitea uses PUT).
"""

import re
from typing import Any

from app.gitea_client import GiteaClient

_LINK_NEXT = re.compile(r'<([^>]+)>;\s*rel="next"')


class GitHubClient(GiteaClient):
    kind = "github"
    api_suffix = ""
    auth_scheme = "Bearer"

    def _extra_headers(self) -> dict:
        return {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

    # ── Pagination (Link headers) ──────────────────────────

    async def _get_all_pages(self, path: str, **params: Any) -> list[dict]:
        items: list[dict] = []
        url: str | None = path
        query: dict[str, Any] = {**params, "per_page": 100}
        pages = 0
        while url and pages < 50:
            pages += 1
            if url.startswith("http"):
                r = await self.client.get(url)
            else:
                r = await self.client.get(url, params=query)
            r.raise_for_status()
            data = r.json()
            if isinstance(data, dict) and "items" in data:
                batch = data["items"]
            elif isinstance(data, list):
                batch = data
            else:
                batch = []
            items.extend(batch)
            match = _LINK_NEXT.search(r.headers.get("Link", ""))
            url = match.group(1) if match else None
            query = {}
        return items

    # ── Repo index ─────────────────────────────────────────

    async def _fetch_search_repos(self) -> list[dict]:
        return await self._get_all_pages(
            "/user/repos", affiliation="owner,collaborator,organization_member"
        )

    # ── Labels: the exposed id IS the name ─────────────────

    async def get_labels(self, owner: str, repo: str) -> list[dict]:
        labels = await super().get_labels(owner, repo)
        for label in labels:
            label["id"] = label["name"]
        return labels

    async def create_label(
        self, owner: str, repo: str, name: str, color: str
    ) -> dict:
        label = await super().create_label(owner, repo, name, color)
        label["id"] = label["name"]
        return label

    async def add_issue_label(
        self, owner: str, repo: str, number: int, label_id
    ) -> None:
        r = await self.client.post(
            f"/repos/{owner}/{repo}/issues/{number}/labels",
            json={"labels": [str(label_id)]},
        )
        r.raise_for_status()

    async def remove_issue_label(
        self, owner: str, repo: str, number: int, label_id
    ) -> None:
        r = await self.client.delete(
            f"/repos/{owner}/{repo}/issues/{number}/labels/{label_id}"
        )
        r.raise_for_status()

    # ── Issue search ───────────────────────────────────────

    async def search_issues(
        self,
        q: str = "",
        state: str = "open",
        assigned_by: str | None = None,
    ) -> list[dict]:
        qualifiers = [q.strip(), "is:issue", f"state:{state}"]
        if assigned_by:
            qualifiers.append(f"assignee:{assigned_by}")
        query = " ".join(part for part in qualifiers if part)
        raw = await self._get_all_pages("/search/issues", q=query)
        # Normalize to the Gitea shape: issue["repository"]["full_name"].
        for item in raw:
            url = item.get("repository_url", "")
            full_name = url.rsplit("/repos/", 1)[-1] if "/repos/" in url else ""
            item["repository"] = {"full_name": full_name}
        return raw

    # ── Archiving (PATCH on GitHub) ────────────────────────

    async def set_archived(
        self, owner: str, repo: str, archived: bool
    ) -> dict:
        r = await self.client.patch(
            f"/repos/{owner}/{repo}", json={"archived": archived}
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()

    # ── Branches (git/refs on GitHub) ──────────────────────

    async def create_branch(
        self, owner: str, repo: str, new_branch: str, old_branch: str
    ) -> dict:
        ref = await self._get(
            f"/repos/{owner}/{repo}/git/ref/heads/{old_branch}"
        )
        sha = (ref.get("object") or {}).get("sha", "")
        r = await self.client.post(
            f"/repos/{owner}/{repo}/git/refs",
            json={"ref": f"refs/heads/{new_branch}", "sha": sha},
        )
        r.raise_for_status()
        return r.json()