challenge_status_tracker/tracker/gitlab.py

"""GitLab REST access: the challenge repos we read, and the one file we write."""
from __future__ import annotations

import base64
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from urllib.parse import quote, urlsplit

import requests


@dataclass
class Repo:
    path: str
    web_url: str
    ssh_url: str
    last_activity: str  # YYYY-MM-DD
    readme: str | None  # None: the repo has no README at all
    id: int = 0
    branch: str | None = None
    # only fetched for challenges whose README claims Testing Ready or beyond
    files: set[str] | None = None  # every file and dir in the repo
    desc: str | None = None        # description.yml, None if there is none


class GitLab:
    """Access to one group, plus the one file we write elsewhere. The token needs
    Project: Read (to list the group) and Repository: Read (to download files);
    --challs additionally needs read + write on the deployment repo."""

    def __init__(self, group_url: str, token: str, timeout: int = 20):
        parts = urlsplit(group_url)
        self.api = f"{parts.scheme}://{parts.netloc}/api/v4"
        self.group = parts.path.strip("/")
        self.timeout = timeout
        self.s = requests.Session()
        self.s.headers["PRIVATE-TOKEN"] = token

    def _get(self, url: str, **params) -> requests.Response:
        return self._req("GET", url, params=params)

    def _req(self, method: str, url: str, ok: tuple = (), **kw) -> requests.Response:
        r = self.s.request(method, url, timeout=self.timeout, **kw)
        if r.status_code in ok:
            return r
        if r.status_code == 403:
            # fine-grained tokens say exactly which permission is missing
            try:
                why = r.json().get("error_description") or r.text
            except ValueError:
                why = r.text
            raise RuntimeError(f"gitlab refused {r.url}: {why}")
        r.raise_for_status()
        return r

    def projects(self) -> list[dict]:
        out, page = [], "1"
        while page:
            r = self._get(f"{self.api}/groups/{quote(self.group, safe='')}/projects",
                          include_subgroups="true", per_page="100", page=page,
                          order_by="path", sort="asc")
            out += r.json()
            page = r.headers.get("X-Next-Page", "")
        # repos GitLab is about to delete are renamed "...-deletion_scheduled-<id>"
        return [p for p in out if not p.get("archived") and not p.get("marked_for_deletion_on")]

    def readme(self, p: dict) -> str | None:
        branch, url = p.get("default_branch"), p.get("readme_url")
        if not branch or not url or f"/-/blob/{branch}/" not in url:
            return None
        return self._raw(p["id"], url.split(f"/-/blob/{branch}/", 1)[1], branch)

    def _raw(self, project_id: int, path: str, branch: str) -> str:
        r = self._get(f"{self.api}/projects/{project_id}/repository/files/"
                      f"{quote(path, safe='')}/raw", ref=branch)
        return r.content.decode("utf-8", "replace")

    def repos(self) -> list[Repo]:
        projects = self.projects()

        def fetch(p: dict) -> Repo:
            return Repo(path=p["path"], web_url=p["web_url"],
                        ssh_url=p["ssh_url_to_repo"],
                        last_activity=(p.get("last_activity_at") or "")[:10],
                        readme=self.readme(p), id=p["id"], branch=p.get("default_branch"))

        with ThreadPoolExecutor(max_workers=8) as pool:
            return list(pool.map(fetch, projects))

    def load_files(self, repo: Repo) -> None:
        """Fill in repo.files and repo.desc, for checks.repo_blockers()."""
        repo.files = set()
        if not repo.branch:
            return
        page = "1"
        while page:
            r = self._get(f"{self.api}/projects/{repo.id}/repository/tree", ref=repo.branch,
                          recursive="true", per_page="100", page=page)
            repo.files |= {t["path"] for t in r.json()}
            page = r.headers.get("X-Next-Page", "")
        if "description.yml" in repo.files:
            repo.desc = self._raw(repo.id, "description.yml", repo.branch)

    def write_file(self, repo_url: str, path: str, content: str, branch: str | None,
                   message) -> bool:
        """Commit `content` to `path` in the repo at `repo_url` (a web URL), creating
        the file if needed; `message(old_content)` makes the commit message.
        False if the file already said exactly this."""
        proj = quote(urlsplit(repo_url).path.strip("/").split("/-/", 1)[0], safe="")
        base = f"{self.api}/projects/{proj}"
        if not branch:
            branch = self._get(base).json()["default_branch"]
        url = f"{base}/repository/files/{quote(path, safe='')}"
        r = self._req("GET", url, ok=(404,), params={"ref": branch})
        exists = r.status_code != 404
        old = base64.b64decode(r.json()["content"]).decode("utf-8", "replace") if exists else ""
        if exists and old == content:
            return False
        self._req("PUT" if exists else "POST", url,
                  json={"branch": branch, "content": content, "commit_message": message(old)})
        return True