github_api_client.py

"""GitHub API client for fetching and managing notifications."""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

import requests

logger = logging.getLogger(__name__)

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


class GitHubAPIClient:
    """Minimal GitHub REST API client focused on notifications."""

    def __init__(self, token: str):
        self._session = requests.Session()
        self._session.headers.update(
            {
                "Authorization": f"token {token}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            }
        )

    def _get(self, path: str, params: Optional[Dict] = None) -> Any:
        resp = self._session.get(f"{GITHUB_API}{path}", params=params, timeout=15)
        resp.raise_for_status()
        return resp.json()

    def _patch(self, path: str, data: Dict) -> Any:
        resp = self._session.patch(f"{GITHUB_API}{path}", json=data, timeout=15)
        resp.raise_for_status()
        return resp.json() if resp.content else {}

    def _delete(self, path: str) -> None:
        resp = self._session.delete(f"{GITHUB_API}{path}", timeout=15)
        resp.raise_for_status()

    def get_notifications(
        self, all_: bool = False, participating: bool = False, per_page: int = 50
    ) -> List[Dict]:
        """Fetch unread (or all) notifications."""
        return self._get(
            "/notifications",
            params={"all": str(all_).lower(), "participating": str(participating).lower(), "per_page": per_page},
        )

    def mark_thread_read(self, thread_id: str) -> None:
        """Mark a notification thread as read."""
        self._patch(f"/notifications/threads/{thread_id}", {})

    def mute_thread(self, thread_id: str) -> None:
        """Mute (unsubscribe) a notification thread."""
        resp = self._session.put(
            f"{GITHUB_API}/notifications/threads/{thread_id}/subscription",
            json={"ignored": True},
            timeout=15,
        )
        resp.raise_for_status()

    def add_label(self, owner: str, repo: str, issue_number: int, label: str) -> None:
        """Add a label to an issue or PR."""
        resp = self._session.post(
            f"{GITHUB_API}/repos/{owner}/{repo}/issues/{issue_number}/labels",
            json={"labels": [label]},
            timeout=15,
        )
        resp.raise_for_status()