82 lines
4 KiB
Python
82 lines
4 KiB
Python
"""Profil (E-Mail, Benachrichtigungen), Übersicht der Zustellung (Plattform-Admin) und Webhook je Workspace."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import HTMLResponse, Response
|
|
from sqlalchemy import select
|
|
|
|
from app import web
|
|
from app.core.errors import Invalid
|
|
from app.identity.admin import require as require_platform_admin
|
|
from app.identity.models import User
|
|
from app.notify import service as notify
|
|
from app.workspaces import service as workspaces
|
|
|
|
Text = Annotated[str, Form()]
|
|
Flag = Annotated[bool, Form()]
|
|
|
|
|
|
def router() -> APIRouter:
|
|
r = APIRouter()
|
|
|
|
def profile_page(request: Request, principal, error: str = "", notice: str = "", status: int = 200):
|
|
settings = web.settings(request)
|
|
with web.db(request).session() as session:
|
|
user = session.get(User, principal.user_id)
|
|
context = {"user": user, "error": error, "notice": notice,
|
|
"mail_ready": bool(settings.smtp_host and settings.smtp_from)}
|
|
return web.render(request, "profile.html", context, status)
|
|
|
|
@r.get("/profile")
|
|
def profile(request: Request, principal: web.User) -> HTMLResponse:
|
|
return profile_page(request, principal)
|
|
|
|
@r.post("/profile")
|
|
def save_profile(request: Request, principal: web.Checked, email: Text = "", notify_me: Flag = False) -> Response:
|
|
try:
|
|
with web.db(request).session() as session:
|
|
notify.save_profile(session, principal.user_id, email, notify_me)
|
|
except Invalid as e:
|
|
return profile_page(request, principal, error=str(e), status=400)
|
|
return profile_page(request, principal, notice="Gespeichert.")
|
|
|
|
@r.post("/profile/test-email")
|
|
def test_email(request: Request, principal: web.Checked) -> Response:
|
|
try:
|
|
with web.db(request).session() as session:
|
|
notify.test_email(session, principal.user_id)
|
|
except Invalid as e:
|
|
return profile_page(request, principal, error=str(e), status=400)
|
|
return profile_page(request, principal, notice="Test-Mail vorgemerkt; der Worker verschickt sie in wenigen "
|
|
"Sekunden. Den Stand zeigt der Plattform-Admin unter "
|
|
"„Benachrichtigungen“.")
|
|
|
|
@r.get("/admin/notifications")
|
|
def overview(request: Request, principal: web.User) -> HTMLResponse:
|
|
require_platform_admin(principal)
|
|
settings = web.settings(request)
|
|
with web.db(request).session() as session:
|
|
items = notify.recent(session)
|
|
names = dict(session.execute(select(User.id, User.display_name).where(
|
|
User.id.in_({i.user_id for i in items if i.user_id}))).all())
|
|
return web.render(request, "admin_notifications.html", {
|
|
"items": items, "names": names, "settings": settings,
|
|
"mail_ready": bool(settings.smtp_host and settings.smtp_from)})
|
|
|
|
@r.post("/workspaces/{workspace_id}/webhook")
|
|
def set_webhook(request: Request, principal: web.Checked, workspace_id: int, url: Text = "",
|
|
remove: Flag = False) -> Response:
|
|
with web.db(request).session() as session:
|
|
workspace = workspaces.require_admin(session, principal, workspace_id).workspace
|
|
notify.set_webhook(session, web.settings(request), web.vault(request), workspace, url, remove)
|
|
return web.redirect(f"/workspaces/{workspace_id}/settings?tab=general&saved=1#notifications")
|
|
|
|
@r.post("/workspaces/{workspace_id}/webhook/test")
|
|
def test_webhook(request: Request, principal: web.Checked, workspace_id: int) -> Response:
|
|
with web.db(request).session() as session:
|
|
workspace = workspaces.require_admin(session, principal, workspace_id).workspace
|
|
notify.test_webhook(session, workspace)
|
|
return web.redirect(f"/workspaces/{workspace_id}/settings?tab=general&saved=1#notifications")
|
|
|
|
return r
|