159 lines
8.4 KiB
Python
159 lines
8.4 KiB
Python
"""Hugin: Host und Übersicht (Plattform-Admin, /admin/hugin); Key, Pakete wählen, abgleichen und
|
||
übernehmen (Workspace-Admin, Reiter „Hugin“ der Einstellungen)."""
|
||
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Form, Request
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from fastapi.responses import HTMLResponse, Response
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app import web
|
||
from app.core.config import Settings
|
||
from app.core.errors import Invalid
|
||
from app.hugin import service as hugin
|
||
from app.hugin.client import HuginError
|
||
from app.identity.admin import require as require_platform_admin
|
||
from app.workspaces import service as workspaces
|
||
from app.workspaces.models import Workspace
|
||
|
||
Text = Annotated[str, Form()]
|
||
|
||
|
||
def tab_context(session: Session, settings: Settings, workspace: Workspace) -> dict:
|
||
"""Für den Reiter „Hugin“ der Workspace-Einstellungen; ohne Aufruf an Hugin."""
|
||
access = hugin.access(session, workspace.id)
|
||
return {"host": hugin.host(session), "access": access,
|
||
"findings": hugin.describe(access.findings or {}) if access else [],
|
||
"subscriptions": hugin.subscriptions(session, workspace.id), "locks": hugin.locks(session, workspace.id),
|
||
"kind_names": hugin.KIND_NAMES}
|
||
|
||
|
||
def router() -> APIRouter:
|
||
r = APIRouter()
|
||
|
||
def back(workspace_id: int) -> Response:
|
||
return web.redirect(f"/workspaces/{workspace_id}/settings?tab=hugin&saved=1")
|
||
|
||
def hugin_page(request: Request, principal, workspace_id: int, template: str, work, status: int = 200):
|
||
"""Seiten mit Aufruf an Hugin (Katalog, Abgleich); Fehler von Hugin erscheinen auf der Seite."""
|
||
settings, vault = web.settings(request), web.vault(request)
|
||
with web.db(request).session() as session:
|
||
workspace = workspaces.require_admin(session, principal, workspace_id).workspace
|
||
context = {"workspace": workspace, "current_workspace": workspace, "error": "",
|
||
"kind_names": hugin.KIND_NAMES, **tab_context(session, settings, workspace)}
|
||
try:
|
||
context.update(work(session, settings, vault))
|
||
except HuginError as e:
|
||
context["error"], status = str(e), 502
|
||
return web.render(request, template, context, status)
|
||
|
||
def admin_page(request: Request, principal, error: str = "", notice: str = "", status: int = 200,
|
||
url: str = "") -> HTMLResponse:
|
||
require_platform_admin(principal)
|
||
with web.db(request).session() as session:
|
||
instance = hugin.host(session)
|
||
context = {"host": instance, "workspaces": hugin.overview(session), "error": error, "notice": notice,
|
||
"url": url or (instance.url if instance else web.settings(request).hugin_url)}
|
||
return web.render(request, "admin_hugin.html", context, status)
|
||
|
||
@r.get("/admin/hugin")
|
||
def admin_overview(request: Request, principal: web.User) -> HTMLResponse:
|
||
return admin_page(request, principal)
|
||
|
||
def check_and_show(request: Request, principal, notice: str) -> HTMLResponse:
|
||
with web.db(request).session() as session:
|
||
row = hugin.check_host(session, principal, web.settings(request))
|
||
ok, message = row.status == "ok", row.status_message
|
||
if ok:
|
||
return admin_page(request, principal, notice=f"{notice} Hugin ist erreichbar.".strip())
|
||
return admin_page(request, principal, error=f"{notice} {message}".strip())
|
||
|
||
@r.post("/admin/hugin")
|
||
def save_host(request: Request, principal: web.Checked, url: Text) -> HTMLResponse:
|
||
try:
|
||
with web.db(request).session() as session:
|
||
hugin.save_host(session, principal, web.settings(request), url)
|
||
except Invalid as e:
|
||
return admin_page(request, principal, error=str(e), status=400, url=url)
|
||
return check_and_show(request, principal, "Gespeichert.")
|
||
|
||
@r.post("/admin/hugin/check")
|
||
async def check_host(request: Request, principal: web.Checked) -> HTMLResponse:
|
||
return await run_in_threadpool(check_and_show, request, principal, "")
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/access")
|
||
async def save_access(request: Request, principal: web.Checked, workspace_id: int, key: Text) -> Response:
|
||
def work():
|
||
with web.db(request).session() as session:
|
||
hugin.save_access(session, principal, web.settings(request), web.vault(request), workspace_id, key)
|
||
await run_in_threadpool(work)
|
||
return back(workspace_id)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/access/test")
|
||
async def test_access(request: Request, principal: web.Checked, workspace_id: int) -> Response:
|
||
def work():
|
||
with web.db(request).session() as session:
|
||
hugin.test_access(session, principal, web.settings(request), web.vault(request), workspace_id)
|
||
await run_in_threadpool(work)
|
||
return back(workspace_id)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/access/remove")
|
||
def remove_access(request: Request, principal: web.Checked, workspace_id: int) -> Response:
|
||
with web.db(request).session() as session:
|
||
hugin.remove_access(session, principal, workspace_id)
|
||
return back(workspace_id)
|
||
|
||
@r.get("/workspaces/{workspace_id}/hugin/catalog")
|
||
async def catalog(request: Request, principal: web.User, workspace_id: int, q: str = "") -> HTMLResponse:
|
||
def work(session, settings, vault):
|
||
packages = hugin.catalog(session, settings, vault, principal, workspace_id)
|
||
if q:
|
||
packages = [p for p in packages if q.lower() in (p["id"] + " " + p.get("description", "")).lower()]
|
||
return {"packages": packages, "q": q,
|
||
"chosen": {s.package_id for s in hugin.subscriptions(session, workspace_id)}}
|
||
return await run_in_threadpool(hugin_page, request, principal, workspace_id, "hugin_catalog.html", work)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/check")
|
||
async def check(request: Request, principal: web.Checked, workspace_id: int) -> HTMLResponse:
|
||
def work(session, settings, vault):
|
||
return {"plan": hugin.plan(session, settings, vault, principal, workspace_id)}
|
||
return await run_in_threadpool(hugin_page, request, principal, workspace_id, "hugin_plan.html", work)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/apply")
|
||
async def apply(request: Request, principal: web.Checked, workspace_id: int) -> Response:
|
||
def work(session, settings, vault):
|
||
return {"plan": hugin.apply(session, settings, vault, principal, workspace_id), "applied": True}
|
||
page = await run_in_threadpool(hugin_page, request, principal, workspace_id, "hugin_plan.html", work)
|
||
return page if page.status_code != 200 else back(workspace_id)
|
||
|
||
@r.get("/workspaces/{workspace_id}/hugin/import")
|
||
async def import_page(request: Request, principal: web.User, workspace_id: int) -> HTMLResponse:
|
||
proposals = await run_in_threadpool(hugin.repo_proposals, web.db(request), web.settings(request),
|
||
web.vault(request), principal, workspace_id)
|
||
|
||
def work(session, settings, vault):
|
||
chosen = {s.package_id: s.spec for s in hugin.subscriptions(session, workspace_id)}
|
||
return {"proposals": proposals, "chosen": chosen}
|
||
return await run_in_threadpool(hugin_page, request, principal, workspace_id, "hugin_import.html", work)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/import")
|
||
def import_packages(request: Request, principal: web.Checked, workspace_id: int,
|
||
package: Annotated[list[str], Form()] = []) -> Response: # noqa: B006 – FastAPI-Formular
|
||
with web.db(request).session() as session:
|
||
hugin.add_many(session, principal, workspace_id, package)
|
||
return back(workspace_id)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/packages")
|
||
def add(request: Request, principal: web.Checked, workspace_id: int, package: Text) -> Response:
|
||
with web.db(request).session() as session:
|
||
hugin.add(session, principal, workspace_id, package)
|
||
return back(workspace_id)
|
||
|
||
@r.post("/workspaces/{workspace_id}/hugin/packages/{package_id}/remove")
|
||
def remove(request: Request, principal: web.Checked, workspace_id: int, package_id: str) -> Response:
|
||
with web.db(request).session() as session:
|
||
hugin.remove(session, principal, workspace_id, package_id)
|
||
return back(workspace_id)
|
||
|
||
return r
|