Thomas Stallinger
43c75df1ed
Neue Sektion auf der /admin/kunden/{id}-Detailseite: Liste bestehender
Bereitstellungsvorlagen der Organisation + Formular zum Anlegen einer
neuen (Workspace, Backend, Root-Dateisystem, Default/Diskverschluesselung/
Secure-Boot), ruft den neuen provisioning-server-Endpoint auf. Schliesst
die in Phase 5 dokumentierte Luecke: bisher hatte eine frische
Kundenorganisation keine Moeglichkeit, ueberhaupt eine Vorlage zu
bekommen, und damit eine leere Auswahl auf der
Installationsmedium-Selfservice-Seite.
Lokal end-to-end verifiziert (siehe provisioning-server-Commit).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
from fastapi import APIRouter, Depends, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from anode_client import anode_request
|
|
from auth import UserCreate, UserManager, get_user_manager, require_admin
|
|
from db import get_database_connection
|
|
from templating import templates
|
|
|
|
|
|
router = APIRouter(prefix="/admin/kunden")
|
|
|
|
|
|
def fetch_benutzer_fuer_organisation(organization_id: str) -> list[dict]:
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT email, is_superuser, created_at FROM benutzer WHERE organization_id = %s ORDER BY email",
|
|
(organization_id,),
|
|
)
|
|
return [
|
|
{"email": email, "is_superuser": is_superuser, "created_at": created_at}
|
|
for email, is_superuser, created_at in cur.fetchall()
|
|
]
|
|
|
|
|
|
@router.get("")
|
|
def index(request: Request, user: dict = Depends(require_admin)):
|
|
result = anode_request("GET", "/api/v1/organizations")
|
|
organizations = result.get("organizations", []) if result.get("success") else []
|
|
|
|
return templates.TemplateResponse(
|
|
request, "admin/kunden_index.html", {"user": user, "organizations": organizations}
|
|
)
|
|
|
|
|
|
@router.post("")
|
|
def organisation_anlegen(name: str = Form(...), user: dict = Depends(require_admin)):
|
|
anode_request("POST", "/api/v1/organizations", json={"name": name})
|
|
|
|
return RedirectResponse("/admin/kunden", status_code=303)
|
|
|
|
|
|
@router.get("/{organization_id}")
|
|
def detail(request: Request, organization_id: str, user: dict = Depends(require_admin)):
|
|
orgs_result = anode_request("GET", "/api/v1/organizations")
|
|
organizations = orgs_result.get("organizations", []) if orgs_result.get("success") else []
|
|
organization = next((o for o in organizations if o["id"] == organization_id), None)
|
|
|
|
codes_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/activation-codes")
|
|
activation_codes = codes_result.get("activation_codes", []) if codes_result.get("success") else []
|
|
|
|
benutzer = fetch_benutzer_fuer_organisation(organization_id)
|
|
|
|
vorlagen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen")
|
|
bereitstellungsvorlagen = vorlagen_result.get("bereitstellungsvorlagen", []) if vorlagen_result.get("success") else []
|
|
|
|
workspaces_result = anode_request("GET", "/api/v1/workspaces")
|
|
workspaces = workspaces_result.get("workspaces", []) if workspaces_result.get("success") else []
|
|
|
|
backends_result = anode_request("GET", "/api/v1/backends")
|
|
backends = backends_result.get("backends", []) if backends_result.get("success") else []
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/kunden_detail.html",
|
|
{
|
|
"user": user,
|
|
"organization": organization,
|
|
"organization_id": organization_id,
|
|
"activation_codes": activation_codes,
|
|
"benutzer": benutzer,
|
|
"bereitstellungsvorlagen": bereitstellungsvorlagen,
|
|
"workspaces": workspaces,
|
|
"backends": backends,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/{organization_id}/bereitstellungsvorlagen")
|
|
def bereitstellungsvorlage_anlegen(
|
|
organization_id: str,
|
|
workspace_id: str = Form(...),
|
|
backend_id: str = Form(...),
|
|
label: str = Form(...),
|
|
is_default: str = Form(default=""),
|
|
disk_encryption: str = Form(default=""),
|
|
root_filesystem: str = Form(default="ext4"),
|
|
secure_boot_required: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
):
|
|
anode_request(
|
|
"POST",
|
|
f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen",
|
|
json={
|
|
"workspace_id": workspace_id,
|
|
"backend_id": backend_id,
|
|
"label": label,
|
|
"is_default": is_default == "on",
|
|
"disk_encryption": disk_encryption == "on",
|
|
"root_filesystem": root_filesystem,
|
|
"secure_boot_required": secure_boot_required == "on",
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/activation-codes")
|
|
def aktivierungscode_anlegen(organization_id: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organizations/{organization_id}/activation-codes")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/konten")
|
|
async def konto_anlegen(
|
|
organization_id: str,
|
|
email: str = Form(...),
|
|
password: str = Form(...),
|
|
is_superuser: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
user_manager: UserManager = Depends(get_user_manager),
|
|
):
|
|
await user_manager.create(
|
|
UserCreate(
|
|
email=email,
|
|
password=password,
|
|
organization_id=organization_id,
|
|
is_superuser=is_superuser == "on",
|
|
)
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|