pr_metadata.py

#!/usr/bin/env python3
"""Load current pull-request metadata, preferring the API over event payloads."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Callable


@dataclass(frozen=True)
class PullRequestMetadata:
    number: int
    body: str
    updated_at: str
    labels: tuple[str, ...]
    source: str


class TransientPRMetadataError(RuntimeError):
    """The current metadata API was unavailable after bounded retries."""


def _parse_metadata(data: dict, source: str) -> PullRequestMetadata:
    labels = data.get("labels") or []
    label_names = tuple(
        sorted(
            label["name"] if isinstance(label, dict) else str(label)
            for label in labels
            if (isinstance(label, str) and label) or (isinstance(label, dict) and label.get("name"))
        )
    )
    number = data.get("number")
    if not isinstance(number, int):
        raise RuntimeError(f"PR metadata from {source} did not include a numeric PR number")
    return PullRequestMetadata(
        number=number,
        body=str(data.get("body") or ""),
        updated_at=str(data.get("updated_at") or data.get("updatedAt") or "unknown"),
        labels=label_names,
        source=source,
    )


# A single flaky GitHub API response must not fail an unrelated PR's required
# preflight, so transient failures receive bounded deterministic retries.
_API_ATTEMPTS = 3
_TRANSIENT_HTTP_STATUSES = frozenset({500, 502, 503, 504})


def load_from_event_file(event_path: Path, expected_number: int) -> PullRequestMetadata:
    """Read the PR snapshot that triggered this workflow after an API outage."""
    try:
        event = json.loads(event_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"could not read GitHub event payload: {exc}") from exc
    if not isinstance(event, dict) or not isinstance(event.get("pull_request"), dict):
        raise RuntimeError("GitHub event payload did not include a pull_request object")
    pull_request = {**event["pull_request"], "number": event.get("number")}
    metadata = _parse_metadata(pull_request, f"GitHub event payload {event_path}")
    if metadata.number != expected_number:
        raise RuntimeError(
            f"GitHub event payload PR number {metadata.number} did not match expected PR #{expected_number}"
        )
    return metadata


def load_from_api(
    repository: str,
    number: int,
    token: str,
    *,
    opener: Callable[..., object] = urllib.request.urlopen,
    sleeper: Callable[[float], None] = time.sleep,
) -> PullRequestMetadata:
    if not token:
        raise RuntimeError("GITHUB_TOKEN is required to read current PR metadata")
    url = f"https://api.github.com/repos/{repository}/pulls/{number}"
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
            "User-Agent": "omi-pr-preflight",
        },
    )
    for attempt in range(1, _API_ATTEMPTS + 1):
        retryable: RuntimeError
        try:
            with opener(request, timeout=15) as response:  # type: ignore[attr-defined]
                payload = json.load(response)
        except urllib.error.HTTPError as exc:
            error = RuntimeError(f"GitHub API returned HTTP {exc.code} while reading PR #{number}")
            exc.close()
            if exc.code not in _TRANSIENT_HTTP_STATUSES:
                raise error from exc
            retryable = error
        except (urllib.error.URLError, OSError, json.JSONDecodeError) as exc:
            reason = getattr(exc, "reason", exc)
            retryable = RuntimeError(f"GitHub API request failed while reading PR #{number}: {reason}")
        else:
            if not isinstance(payload, dict):
                raise RuntimeError(f"GitHub API returned invalid metadata for PR #{number}")
            return _parse_metadata(payload, f"GitHub API PR #{number}")
        if attempt == _API_ATTEMPTS:
            raise TransientPRMetadataError(str(retryable)) from retryable
        sleeper(2.0 * attempt)
    raise AssertionError("unreachable")


def load_from_gh(root: Path) -> PullRequestMetadata:
    try:
        result = subprocess.run(
            ["gh", "pr", "view", "--json", "number,body,updatedAt,labels"],
            cwd=root,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError("gh is not installed; pass --pr-body-file before the first push") from exc
    except subprocess.CalledProcessError as exc:
        detail = exc.stderr.strip() or "no pull request is associated with this branch"
        raise RuntimeError(f"could not discover the current PR with gh: {detail}") from exc
    payload = json.loads(result.stdout)
    if not isinstance(payload, dict):
        raise RuntimeError("gh returned invalid PR metadata")
    return _parse_metadata(payload, "gh current PR")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repository", required=True, help="GitHub repository as owner/name")
    parser.add_argument("--pr-number", required=True, type=int)
    parser.add_argument("--output", required=True, type=Path, help="File to receive the current PR body")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        metadata = load_from_api(args.repository, args.pr_number, os.getenv("GITHUB_TOKEN", ""))
    except RuntimeError as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        return 1
    args.output.write_text(metadata.body, encoding="utf-8")
    print(f"Loaded {metadata.source}, updated_at={metadata.updated_at}.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())