119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
"""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 resolve_document_id(self, document_id_or_slug: str) -> str:
|
|
"""Löst eine Dokument-ID oder Friendly-URL-Slug in die echte UUID auf."""
|
|
data = self._post("/documents.info", {"id": document_id_or_slug})
|
|
return data["data"]["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 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
|