92 lines
3.8 KiB
Python
92 lines
3.8 KiB
Python
"""
|
|
Kleiner HTTP-Client fuer die Forge-Contents-API (phase-008-itsm-repo-audit).
|
|
|
|
Analog zum bestehenden Muster in app.py (urllib.request statt einer externen
|
|
HTTP-Bibliothek -- keine zusaetzliche Abhaengigkeit fuer nur zwei Aufrufe).
|
|
Spricht ausschliesslich die Forge-REST-API (siehe mscadm/forge
|
|
forge-web/src/api.rs), die bewusst Gitea-API-kompatibel gehalten ist.
|
|
|
|
ENV (siehe auch app.py-Docstring):
|
|
FORGE_BASE_URL Basis-URL des Forge-Servers, z.B. http://127.0.0.1:8095
|
|
FORGE_SERVICE_TOKEN API-Token eines Forge-Nutzers mit role=admin (siehe
|
|
Forge /api/v1/admin/users) -- dient hier NICHT der
|
|
Admin-API, sondern normalen Contents-Schreibzugriffen;
|
|
admin ist ausreichend, da Forge (noch) kein feineres
|
|
Repo-Schreibrecht kennt.
|
|
|
|
Bewusster Scope-Schnitt: nur get_contents/update_contents (was diese Phase
|
|
tatsaechlich braucht), keine generische Forge-API-Client-Klasse mit allen
|
|
Endpunkten -- analog zur Begruendung in forge-web/src/api.rs ("nicht 100%
|
|
Paritaet ab Tag 1").
|
|
"""
|
|
import base64
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
FORGE_BASE_URL = os.getenv("FORGE_BASE_URL", "").rstrip("/")
|
|
FORGE_SERVICE_TOKEN = os.getenv("FORGE_SERVICE_TOKEN", "")
|
|
|
|
|
|
class ForgeClientError(Exception):
|
|
"""Fehler beim Sprechen mit der Forge-API -- Aufrufer soll dies dem
|
|
Nutzer als Fehlermeldung zeigen, nicht stillschweigend schlucken."""
|
|
|
|
|
|
def _request(method, path, body=None):
|
|
if not FORGE_BASE_URL:
|
|
raise ForgeClientError("FORGE_BASE_URL ist nicht konfiguriert")
|
|
url = FORGE_BASE_URL + path
|
|
headers = {"Authorization": "token " + FORGE_SERVICE_TOKEN}
|
|
data = None
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
detail = ""
|
|
try:
|
|
detail = e.read().decode("utf-8")
|
|
except Exception:
|
|
pass
|
|
raise ForgeClientError("Forge-API-Fehler (%s): %s" % (e.code, detail or e.reason))
|
|
except urllib.error.URLError as e:
|
|
raise ForgeClientError("Forge nicht erreichbar: %s" % e.reason)
|
|
|
|
|
|
def get_contents(repo, path, ref=None):
|
|
"""Liest eine Datei aus einem Forge-Repo. Rueckgabe: (text, sha).
|
|
`text` ist bereits UTF-8-dekodiert (Repo-Dateien fuer diese Funktion sind
|
|
Text-Dateien wie Doku/Config -- Binaerdateien werden bewusst nicht
|
|
unterstuetzt, siehe project.yaml phase-008 Scope)."""
|
|
q = "?ref=" + urllib.parse.quote(ref, safe="") if ref else ""
|
|
result = _request("GET", "/api/v1/repos/x/%s/contents/%s%s" % (repo, path, q))
|
|
if result.get("type") != "file":
|
|
raise ForgeClientError("Pfad ist keine Datei: %s" % path)
|
|
raw = base64.b64decode(result["content"])
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise ForgeClientError("Datei ist keine UTF-8-Textdatei -- ueber ITSM nicht editierbar")
|
|
return text, result["sha"]
|
|
|
|
|
|
def update_contents(repo, path, content_text, sha, branch, message, author_name, author_email):
|
|
"""Schreibt eine Datei in ein Forge-Repo (optimistisches Sha-Locking wie
|
|
von Forges Contents-API verlangt). Rueckgabe: commit_sha (str)."""
|
|
encoded = base64.b64encode(content_text.encode("utf-8")).decode("ascii")
|
|
body = {
|
|
"content": encoded,
|
|
"sha": sha,
|
|
"branch": branch,
|
|
"message": message,
|
|
"author": {"name": author_name, "email": author_email},
|
|
}
|
|
result = _request("PUT", "/api/v1/repos/x/%s/contents/%s" % (repo, path), body)
|
|
return result["commit"]["sha"]
|