pr-reviewer/app/github_client.py

"""GitHub API client.

Auth resolution order:
  1. GITHUB_TOKEN from config/env
  2. `gh auth token` from the local gh CLI

Read paths: search PRs, fetch PR detail + diff.
Write path: submit a PR review (guarded by an explicit call from the API layer,
which is only reached after the user clicks Post in the dashboard).
"""

from __future__ import annotations

import subprocess
from typing import Any

import httpx2

from .config import Config

API = "https://api.github.com"


class GitHubError(RuntimeError):
    pass


def _raise_for_status(r: httpx2.Response) -> None:
    """Turn a non-2xx GitHub response into a GitHubError with a readable
    message, so the dashboard shows 'GitHub 401: Bad credentials' instead of
    a bare 500."""
    if r.status_code >= 400:
        try:
            msg = r.json().get("message", r.text)
        except Exception:
            msg = r.text
        raise GitHubError(f"GitHub {r.status_code}: {msg}")


def _resolve_token(cfg: Config) -> str:
    if cfg.GITHUB_TOKEN:
        return cfg.GITHUB_TOKEN
    try:
        out = subprocess.run(
            ["gh", "auth", "token"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if out.returncode == 0 and out.stdout.strip():
            return out.stdout.strip()
    except (FileNotFoundError, subprocess.SubprocessError):
        pass
    raise GitHubError(
        "No GitHub token. Set GITHUB_TOKEN in .env or authenticate with `gh auth login`."
    )


class GitHubClient:
    def __init__(self, cfg: Config):
        self.cfg = cfg
        self._token = _resolve_token(cfg)
        self._headers = {
            "Authorization": f"Bearer {self._token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

    # --- identity ---
    def whoami(self) -> str:
        with httpx2.Client(timeout=20) as c:
            r = c.get(f"{API}/user", headers=self._headers)
            _raise_for_status(r)
            return r.json()["login"]

    # --- read: search the review queue ---
    def search_prs(self) -> list[dict[str, Any]]:
        params = {"q": self.cfg.PR_SEARCH_FILTER, "per_page": self.cfg.MAX_PRS}
        with httpx2.Client(timeout=30) as c:
            r = c.get(f"{API}/search/issues", headers=self._headers, params=params)
            _raise_for_status(r)
            items = r.json().get("items", [])
        prs = []
        for it in items:
            # search/issues returns PRs with a pull_request key; parse owner/repo/number from url
            repo_url = it["repository_url"]  # .../repos/{owner}/{repo}
            owner, repo = repo_url.split("/repos/")[1].split("/")
            prs.append(
                {
                    "owner": owner,
                    "repo": repo,
                    "number": it["number"],
                    "title": it["title"],
                    "url": it["html_url"],
                    "author": (it.get("user") or {}).get("login", ""),
                    "updated_at": it["updated_at"],
                    "body": it.get("body") or "",
                    "draft": it.get("draft", False),
                }
            )
        return prs

    # --- read: PR diff ---
    def pr_diff(self, owner: str, repo: str, number: int) -> str:
        headers = dict(self._headers)
        headers["Accept"] = "application/vnd.github.v3.diff"
        with httpx2.Client(timeout=60) as c:
            r = c.get(
                f"{API}/repos/{owner}/{repo}/pulls/{number}",
                headers=headers,
            )
            _raise_for_status(r)
            return r.text

    # --- write: submit a review (side-effectful; called only on user action) ---
    def submit_review(
        self,
        owner: str,
        repo: str,
        number: int,
        body: str,
        event: str,
    ) -> dict[str, Any]:
        event = event.upper()
        if event not in {"COMMENT", "APPROVE", "REQUEST_CHANGES"}:
            raise GitHubError(f"Invalid event: {event}")
        payload = {"body": body, "event": event}
        with httpx2.Client(timeout=30) as c:
            r = c.post(
                f"{API}/repos/{owner}/{repo}/pulls/{number}/reviews",
                headers=self._headers,
                json=payload,
            )
            _raise_for_status(r)
            return r.json()