diet-guard/diet_guard/_sync_github.py

"""Minimal GitHub Contents API client for diet_guard's dumb-storage sync.

GitHub is used purely as file storage via the REST Contents API, not a git
clone -- ported in spirit from ``~/todo``'s sync transport.  There is no
working tree and no git-level merge; the only merge is the domain-level one
in :mod:`diet_guard._sync_merge`.
"""

from __future__ import annotations

import base64
import logging

import requests

from diet_guard._constants import SYNC_TIMEOUT_SECONDS

_logger = logging.getLogger(__name__)

_API_BASE = "https://api.github.com"
_HTTP_NOT_FOUND = 404


class GitHubSyncError(Exception):
    """Raised for a GitHub API failure the caller must not silently ignore."""


class RepoNotFoundError(GitHubSyncError):
    """Raised when the configured repo itself is unreachable.

    Distinguished from a path-404 (nothing pushed to that path yet, which is
    benign -- it just means no other device has synced before) so the caller
    can tell "the repo name is wrong or the PAT isn't scoped to it" apart
    from "no other device has synced yet".
    """


class GitHubSyncClient:
    """Thin wrapper around the subset of the Contents API sync needs."""

    def __init__(self, owner: str, repo: str, token: str) -> None:
        """Create a client scoped to one repo, authenticated with ``token``.

        Args:
            owner: The repo owner/org (e.g. ``"kuhyx"``).
            repo: The repo name (e.g. ``"diet-guard-sync"``).
            token: A GitHub PAT with contents read/write on that repo.
        """
        self._owner = owner
        self._repo = repo
        self._headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
        }

    def _contents_url(self, path: str) -> str:
        return f"{_API_BASE}/repos/{self._owner}/{self._repo}/contents/{path}"

    def _get(self, path: str) -> requests.Response:
        try:
            return requests.get(
                self._contents_url(path),
                headers=self._headers,
                timeout=SYNC_TIMEOUT_SECONDS,
            )
        except requests.RequestException as exc:
            msg = f"network error reading {path}"
            raise GitHubSyncError(msg) from exc

    def _repo_exists(self) -> bool:
        try:
            response = requests.get(
                f"{_API_BASE}/repos/{self._owner}/{self._repo}",
                headers=self._headers,
                timeout=SYNC_TIMEOUT_SECONDS,
            )
        except requests.RequestException:
            return False
        return response.ok

    def _raise_for_missing_path(self, path: str) -> None:
        """Raise :class:`RepoNotFoundError` only if the repo is unreachable.

        A 404 on a path within a reachable repo just means nothing has been
        pushed there yet, which is not an error worth raising on.
        """
        if not self._repo_exists():
            msg = (
                f"{self._owner}/{self._repo} not found, private without "
                f"access, or the token lacks contents permission "
                f"(while reading {path})"
            )
            raise RepoNotFoundError(msg)

    def get_file_text(self, path: str) -> str | None:
        """Return the decoded text content at ``path``, or None if unused.

        Args:
            path: A repo-relative file path, e.g. ``"devices/pc/food_log.json"``.

        Returns:
            The file's text content, or None if nothing has been pushed
            there yet (but the repo itself is reachable).

        Raises:
            RepoNotFoundError: If the repo itself is unreachable.
            GitHubSyncError: For any other non-2xx response or network error.
        """
        response = self._get(path)
        if response.status_code == _HTTP_NOT_FOUND:
            self._raise_for_missing_path(path)
            return None
        if not response.ok:
            msg = f"GET {path} failed: {response.status_code}"
            raise GitHubSyncError(msg)
        data = response.json()
        content = data.get("content", "") if isinstance(data, dict) else ""
        return base64.b64decode(content).decode("utf-8")

    def _existing_sha(self, path: str) -> str | None:
        response = self._get(path)
        if response.status_code == _HTTP_NOT_FOUND:
            self._raise_for_missing_path(path)
            return None
        if not response.ok:
            msg = f"GET {path} (for sha) failed: {response.status_code}"
            raise GitHubSyncError(
                msg,
            )
        data = response.json()
        sha = data.get("sha") if isinstance(data, dict) else None
        return sha if isinstance(sha, str) else None

    def put_file_text(self, path: str, text: str, *, message: str) -> None:
        """Create or update the file at ``path`` with ``text``.

        Args:
            path: A repo-relative file path.
            text: The full new content (this device's complete merged log).
            message: The commit message for this push.

        Raises:
            GitHubSyncError: On any non-2xx response or network error.
        """
        sha = self._existing_sha(path)
        payload: dict[str, object] = {
            "message": message,
            "content": base64.b64encode(text.encode("utf-8")).decode("ascii"),
        }
        if sha is not None:
            payload["sha"] = sha
        try:
            response = requests.put(
                self._contents_url(path),
                headers=self._headers,
                json=payload,
                timeout=SYNC_TIMEOUT_SECONDS,
            )
        except requests.RequestException as exc:
            msg = f"network error pushing {path}"
            raise GitHubSyncError(msg) from exc
        if not response.ok:
            msg = f"PUT {path} failed: {response.status_code}"
            raise GitHubSyncError(msg)

    def list_directory(self, path: str) -> list[str]:
        """Return the entry names directly under ``path`` (empty if unused).

        Args:
            path: A repo-relative directory path, e.g. ``"devices"``.

        Raises:
            RepoNotFoundError: If the repo itself is unreachable.
            GitHubSyncError: For any other non-2xx response or network error.
        """
        response = self._get(path)
        if response.status_code == _HTTP_NOT_FOUND:
            self._raise_for_missing_path(path)
            return []
        if not response.ok:
            msg = f"GET {path} (list) failed: {response.status_code}"
            raise GitHubSyncError(msg)
        data = response.json()
        if not isinstance(data, list):
            return []
        return [
            item["name"]
            for item in data
            if isinstance(item, dict) and isinstance(item.get("name"), str)
        ]