Thomas Stallinger
b44096b48f
routers/admin_kunden.py mit echter Funktionalität: Organisationen auflisten/anlegen, pro Organisation Aktivierungscodes erzeugen und Kundenkonten anlegen (direkt in der Kundenplattform-eigenen benutzer-Tabelle, kein anode-Aufruf nötig). Löst das bisherige manuelle create_user.py-Skript für den Alltagsgebrauch ab. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from uuid import uuid4
|
|
|
|
import bcrypt
|
|
from fastapi import APIRouter, Depends, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from anode_client import anode_request
|
|
from auth import 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, ist_admin, created_at FROM benutzer WHERE organization_id = %s ORDER BY email",
|
|
(organization_id,),
|
|
)
|
|
return [
|
|
{"email": email, "ist_admin": ist_admin, "created_at": created_at}
|
|
for email, ist_admin, 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)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/kunden_detail.html",
|
|
{
|
|
"user": user,
|
|
"organization": organization,
|
|
"organization_id": organization_id,
|
|
"activation_codes": activation_codes,
|
|
"benutzer": benutzer,
|
|
},
|
|
)
|
|
|
|
|
|
@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")
|
|
def konto_anlegen(
|
|
organization_id: str,
|
|
email: str = Form(...),
|
|
password: str = Form(...),
|
|
ist_admin: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
):
|
|
password_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO benutzer (id, organization_id, email, password_hash, ist_admin)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
""",
|
|
(uuid4(), organization_id, email, password_hash, ist_admin == "on"),
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|