cf7ea8f9a6
GrampsWeb expects JSON for the token endpoint, not form-encoded data. Discovered during live API testing against production. Co-Authored-By: Paperclip <noreply@paperclip.ing>
117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
"""
|
||
HTTP-Client für die GrampsWeb REST API.
|
||
|
||
Konfiguration über Umgebungsvariablen:
|
||
GRAMPS_URL – Basis-URL (z.B. http://grampsweb:5000)
|
||
GRAMPS_USERNAME – Benutzername für Login
|
||
GRAMPS_PASSWORD – Passwort für Login
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
|
||
import requests
|
||
|
||
logger = logging.getLogger("gramps_mcp_server.client")
|
||
|
||
|
||
class GrampsWebClient:
|
||
"""HTTP-Client für GrampsWeb mit automatischem Token-Management."""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str | None = None,
|
||
username: str | None = None,
|
||
password: str | None = None,
|
||
):
|
||
self.base_url = (base_url or os.environ.get("GRAMPS_URL", "http://grampsweb:5000")).rstrip("/")
|
||
self.username = username or os.environ.get("GRAMPS_USERNAME", "")
|
||
self.password = password or os.environ.get("GRAMPS_PASSWORD", "")
|
||
self._session = requests.Session()
|
||
self._token: str | None = None
|
||
|
||
def _ensure_auth(self) -> None:
|
||
"""Login falls noch kein Token vorhanden."""
|
||
if self._token:
|
||
return
|
||
if not self.username or not self.password:
|
||
raise RuntimeError(
|
||
"GRAMPS_USERNAME und GRAMPS_PASSWORD müssen gesetzt sein."
|
||
)
|
||
self._login()
|
||
|
||
def _login(self) -> None:
|
||
"""Authentifizierung bei GrampsWeb und Token-Speicherung."""
|
||
login_endpoints = [
|
||
("/api/token/", "json"),
|
||
("/api/login/", "json"),
|
||
]
|
||
for path, mode in login_endpoints:
|
||
url = f"{self.base_url}{path}"
|
||
payload = {"username": self.username, "password": self.password}
|
||
try:
|
||
if mode == "json":
|
||
r = self._session.post(url, json=payload, timeout=15)
|
||
else:
|
||
r = self._session.post(url, data=payload, timeout=15)
|
||
if r.status_code in (200, 201):
|
||
data = r.json()
|
||
token = (
|
||
data.get("access_token")
|
||
or data.get("token")
|
||
or data.get("access")
|
||
)
|
||
if token:
|
||
self._token = token
|
||
self._session.headers["Authorization"] = f"Bearer {token}"
|
||
logger.info("GrampsWeb Login erfolgreich via %s", path)
|
||
return
|
||
except Exception:
|
||
continue
|
||
raise RuntimeError(
|
||
f"GrampsWeb Login fehlgeschlagen. URL: {self.base_url}, User: {self.username}"
|
||
)
|
||
|
||
def _request(self, method: str, path: str, **kwargs) -> requests.Response:
|
||
"""HTTP-Request mit automatischer Re-Auth bei 401."""
|
||
self._ensure_auth()
|
||
kwargs.setdefault("timeout", 30)
|
||
url = f"{self.base_url}{path}"
|
||
r = self._session.request(method, url, **kwargs)
|
||
if r.status_code == 401:
|
||
self._token = None
|
||
self._login()
|
||
r = self._session.request(method, url, **kwargs)
|
||
r.raise_for_status()
|
||
return r
|
||
|
||
def get(self, path: str, **kwargs) -> dict | list:
|
||
"""GET-Request, gibt JSON zurück."""
|
||
return self._request("GET", path, **kwargs).json()
|
||
|
||
def get_raw(self, path: str, **kwargs) -> bytes:
|
||
"""GET-Request, gibt Rohdaten zurück (z.B. für Datei-Downloads)."""
|
||
return self._request("GET", path, **kwargs).content
|
||
|
||
def post(self, path: str, **kwargs) -> dict | list:
|
||
"""POST-Request, gibt JSON zurück."""
|
||
return self._request("POST", path, **kwargs).json()
|
||
|
||
def put(self, path: str, **kwargs) -> dict | list:
|
||
"""PUT-Request, gibt JSON zurück."""
|
||
return self._request("PUT", path, **kwargs).json()
|
||
|
||
|
||
# Singleton-Instanz
|
||
_client: GrampsWebClient | None = None
|
||
|
||
|
||
def get_client() -> GrampsWebClient:
|
||
"""Gibt die (gecachte) Client-Instanz zurück."""
|
||
global _client
|
||
if _client is None:
|
||
_client = GrampsWebClient()
|
||
return _client
|