phpipam-outline-sync/app/outline_client.py

"""Kleiner Client für die Outline API.

Doku: https://www.getoutline.com/developers
"""
from __future__ import annotations

import httpx

from .config import settings


class OutlineClient:
    def __init__(self) -> None:
        self._base = f"{settings.outline_base_url.rstrip('/')}/api"
        self._client = httpx.Client(
            verify=settings.outline_verify_ssl,
            timeout=30.0,
            headers={
                "Authorization": f"Bearer {settings.outline_api_token}",
                "Content-Type": "application/json",
            },
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "OutlineClient":
        return self

    def __exit__(self, *exc) -> None:
        self.close()

    def _post(self, path: str, json: dict) -> dict:
        resp = self._client.post(f"{self._base}{path}", json=json)
        resp.raise_for_status()
        data = resp.json()
        if "data" not in data:
            raise RuntimeError(f"Unerwartete Outline-Antwort von {path}: {data}")
        return data

    def resolve_collection_id(self, collection_id_or_slug: str) -> str:
        """Löst eine Collection-ID oder Friendly-URL-Slug in die echte UUID auf.

        `documents.search`/`documents.create` akzeptieren für `collectionId`
        nur eine vollständige UUID. Die ID in der Browser-Adresszeile ist bei
        neueren Outline-Versionen aber die gekürzte Friendly-URL-ID, keine
        UUID. `collections.info` akzeptiert beides, daher lösen wir hier
        einmal auf.
        """
        data = self._post("/collections.info", {"id": collection_id_or_slug})
        return data["data"]["id"]

    def get_document(self, document_id_or_slug: str) -> dict:
        """Lädt die vollständigen Metadaten eines Dokuments per ID oder
        Friendly-URL-Slug (u.a. für den aktuellen Titel gebraucht)."""
        data = self._post("/documents.info", {"id": document_id_or_slug})
        return data["data"]

    def resolve_document_id(self, document_id_or_slug: str) -> str:
        """Löst eine Dokument-ID oder Friendly-URL-Slug in die echte UUID auf."""
        return self.get_document(document_id_or_slug)["id"]

    def find_document_by_title(
        self, title: str, collection_id: str, parent_document_id: str | None = None
    ) -> dict | None:
        """Sucht ein bestehendes Dokument mit exakt diesem Titel in der Collection.

        Ist parent_document_id gesetzt, wird zusätzlich geprüft, dass der Treffer
        auch wirklich unter dieser Elternseite hängt (z.B. "Netze"), damit gleich
        benannte Dokumente an anderer Stelle in der Collection nicht versehentlich
        überschrieben werden.
        """
        data = self._post(
            "/documents.search",
            {"query": title, "collectionId": collection_id, "limit": 25},
        )
        for hit in data["data"]:
            doc = hit.get("document", hit)
            if doc.get("title") != title:
                continue
            if parent_document_id and doc.get("parentDocumentId") != parent_document_id:
                continue
            return doc
        return None

    def create_document(
        self, title: str, text: str, collection_id: str, parent_document_id: str | None = None
    ) -> dict:
        payload = {
            "title": title,
            "text": text,
            "collectionId": collection_id,
            "publish": True,
        }
        if parent_document_id:
            payload["parentDocumentId"] = parent_document_id
        data = self._post("/documents.create", payload)
        return data["data"]

    def update_document(self, document_id: str, title: str, text: str) -> dict:
        data = self._post(
            "/documents.update",
            {"id": document_id, "title": title, "text": text},
        )
        return data["data"]

    def list_documents(self, collection_id: str, parent_document_id: str | None = None) -> list[dict]:
        """Listet alle Dokumente einer Collection (optional gefiltert auf
        direkte Kinder von parent_document_id), paginiert."""
        docs: list[dict] = []
        offset = 0
        limit = 100
        while True:
            payload: dict = {"collectionId": collection_id, "limit": limit, "offset": offset}
            if parent_document_id:
                payload["parentDocumentId"] = parent_document_id
            data = self._post("/documents.list", payload)
            batch = data["data"]
            docs.extend(batch)
            if len(batch) < limit:
                break
            offset += limit
        return docs

    def list_revisions(self, document_id: str) -> list[dict]:
        """Listet alle Revisionen (Versionsverlauf) eines Dokuments, paginiert."""
        revisions: list[dict] = []
        offset = 0
        limit = 100
        while True:
            data = self._post(
                "/revisions.list",
                {"documentId": document_id, "limit": limit, "offset": offset},
            )
            batch = data["data"]
            revisions.extend(batch)
            if len(batch) < limit:
                break
            offset += limit
        return revisions

    def delete_revision(self, revision_id: str) -> None:
        self._post("/revisions.delete", {"id": revision_id})

    def purge_document_history(self, document_id: str) -> int:
        """Löscht ALLE Revisionen eines Dokuments dauerhaft und unwiederbringlich
        (nicht der aktuelle Inhalt selbst, nur der Versionsverlauf).

        Returns: Anzahl gelöschter Revisionen.
        """
        revisions = self.list_revisions(document_id)
        for revision in revisions:
            self.delete_revision(revision["id"])
        return len(revisions)

    def prune_document_history(self, document_id: str, keep: int) -> int:
        """Behält nur die `keep` neuesten Revisionen eines Dokuments, alle
        älteren werden dauerhaft gelöscht und sind nicht wiederherstellbar.

        Returns: Anzahl gelöschter Revisionen.
        """
        if keep < 0:
            keep = 0
        revisions = self.list_revisions(document_id)
        revisions.sort(key=lambda r: r.get("createdAt") or "", reverse=True)
        to_delete = revisions[keep:]
        for revision in to_delete:
            self.delete_revision(revision["id"])
        return len(to_delete)

    def upsert_document(
        self,
        title: str,
        text: str,
        collection_id: str,
        parent_document_id: str | None = None,
    ) -> tuple[dict, bool]:
        """Legt ein Dokument an oder aktualisiert es, falls der Titel bereits existiert.

        Returns: (document, created)
        """
        existing = self.find_document_by_title(title, collection_id, parent_document_id)
        if existing:
            doc = self.update_document(existing["id"], title, text)
            return doc, False
        doc = self.create_document(title, text, collection_id, parent_document_id)
        return doc, True