claude-regelwerk/.claude/skills/plane-abgleich/scripts/plane/api.py

"""HTTP-Zugriff auf die Plane-API."""

from __future__ import annotations

import time
from typing import Any

import requests

from .konfiguration import Konfiguration

HOECHSTE_WARTEZEIT = 65.0
VERSUCHE_BEI_DROSSELUNG = 3


class ApiFehler(Exception):
    """Plane antwortet nicht wie erwartet."""


class PlaneApi:
    """Kapselt alle Zugriffe auf die Plane-API.

    Es gibt bewusst je erlaubtem HTTP-Verb eine eigene Methode. Ein Verb als
    Parameter durchzureichen würde löschende Aufrufe formulierbar machen.
    """

    def __init__(self, konfiguration: Konfiguration, sitzung=None, schlaeft=time.sleep):
        self._k = konfiguration
        self._s = sitzung if sitzung is not None else requests.Session()
        self._schlaeft = schlaeft

    @property
    def basis(self) -> str:
        return f"{self._k.basis_url}/api/v1/workspaces/{self._k.workspace}"

    def _kopf(self) -> dict[str, str]:
        return {"X-API-Key": self._k.api_key, "Content-Type": "application/json"}

    def _sende(self, verb_funktion, url: str, **kw):
        antwort = None
        for _ in range(VERSUCHE_BEI_DROSSELUNG):
            antwort = verb_funktion(url, headers=self._kopf(), timeout=30, **kw)
            if antwort.status_code != 429:
                break
            self._schlaeft(self._wartezeit(antwort))
        return self._geprueft(antwort)

    def _geprueft(self, antwort):
        if antwort.status_code == 401:
            raise ApiFehler(
                "Plane lehnt den Schlüssel ab (401). PLANE_API_KEY prüfen."
            )
        if antwort.status_code == 403:
            raise ApiFehler(
                f"Plane verweigert den Zugriff (403). Entweder ist der Workspace-Slug "
                f"„{self._k.workspace}“ falsch oder der Schlüssel hat dort "
                f"keine Rechte — beides antwortet gleich."
            )
        if antwort.status_code == 429:
            raise ApiFehler(
                "Plane drosselt weiterhin (429), auch nach mehreren Versuchen. "
                "Später erneut versuchen."
            )
        if antwort.status_code >= 400:
            raise ApiFehler(
                f"Plane antwortet {antwort.status_code}: {antwort.text[:300]}"
            )
        return antwort

    @staticmethod
    def _wartezeit(antwort, jetzt: float | None = None) -> float:
        """Liest X-RateLimit-Reset; der Wert kann Zeitstempel oder Restdauer sein."""
        jetzt = time.time() if jetzt is None else jetzt
        roh = antwort.headers.get("X-RateLimit-Reset", "")
        try:
            wert = float(roh)
        except (TypeError, ValueError):
            return 5.0
        rest = wert - jetzt if wert > 1_000_000 else wert
        return min(max(rest, 1.0), HOECHSTE_WARTEZEIT)

    def _get(self, url: str, **kw):
        return self._sende(self._s.get, url, **kw)

    def _post(self, url: str, **kw):
        return self._sende(self._s.post, url, **kw)

    def _patch(self, url: str, **kw):
        return self._sende(self._s.patch, url, **kw)

    def _alle(self, url: str, params: dict[str, Any] | None = None) -> list[dict]:
        """Folgt der Cursor-Pagination bis zum Ende."""
        params = dict(params or {})
        params.setdefault("per_page", 100)
        gesammelt: list[dict] = []
        while True:
            daten = self._get(url, params=params).json()
            if isinstance(daten, list):
                return daten
            gesammelt.extend(daten.get("results", []))
            if not daten.get("next_page_results"):
                return gesammelt
            cursor = daten.get("next_cursor")
            if not cursor:
                return gesammelt
            params["cursor"] = cursor

    # --- Fachmethoden -----------------------------------------------------

    def projekte(self) -> list[dict]:
        return self._alle(f"{self.basis}/projects/")

    def projekt_anlegen(self, name: str, kennung: str) -> dict:
        gesaeubert = "".join(z for z in kennung.upper() if z.isascii() and z.isalnum())
        if not gesaeubert:
            raise ApiFehler(
                f"Die Kennung „{kennung}“ enthält kein einziges "
                f"verwendbares Zeichen (erlaubt sind A-Z und 0-9)."
            )
        return self._post(
            f"{self.basis}/projects/",
            json={"name": name, "identifier": gesaeubert},
        ).json()

    def states(self, projekt_id: str) -> list[dict]:
        return self._alle(f"{self.basis}/projects/{projekt_id}/states/")

    def work_items(self, projekt_id: str) -> list[dict]:
        return self._alle(f"{self.basis}/projects/{projekt_id}/work-items/")

    def work_item_anlegen(self, projekt_id: str, titel: str, state_id: str) -> dict:
        return self._post(
            f"{self.basis}/projects/{projekt_id}/work-items/",
            json={"name": titel, "state": state_id},
        ).json()

    def work_item_aendern(
        self, projekt_id: str, work_item_id: str, felder: dict[str, Any]
    ) -> dict:
        return self._patch(
            f"{self.basis}/projects/{projekt_id}/work-items/{work_item_id}/",
            json=felder,
        ).json()