feat: add admin user management
This commit is contained in:
@@ -8,14 +8,21 @@ from flask_login import current_user, login_required
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
from ..extensions import csrf, db
|
||||
from ..forms import SettingsProfileForm
|
||||
from ..models import BadgeDefinition, PushSubscription
|
||||
from ..forms import AdminUserForm, SettingsProfileForm
|
||||
from ..models import BadgeDefinition, MonthlyScoreSnapshot, NotificationLog, PushSubscription, TaskInstance, TaskTemplate, User
|
||||
from ..services.notifications import push_enabled
|
||||
|
||||
|
||||
bp = Blueprint("settings", __name__, url_prefix="/settings")
|
||||
|
||||
|
||||
def _require_admin():
|
||||
if not current_user.is_admin:
|
||||
flash("Dieser Bereich ist nur für Admins verfügbar.", "error")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _save_avatar(file_storage) -> str:
|
||||
filename = secure_filename(file_storage.filename or "")
|
||||
ext = Path(filename).suffix.lower() or ".png"
|
||||
@@ -30,6 +37,7 @@ def _save_avatar(file_storage) -> str:
|
||||
@login_required
|
||||
def index():
|
||||
form = SettingsProfileForm(original_email=current_user.email, obj=current_user)
|
||||
admin_form = AdminUserForm(prefix="admin")
|
||||
if form.validate_on_submit():
|
||||
current_user.name = form.name.data.strip()
|
||||
current_user.email = form.email.data.lower().strip()
|
||||
@@ -48,7 +56,9 @@ def index():
|
||||
return render_template(
|
||||
"settings/index.html",
|
||||
form=form,
|
||||
admin_form=admin_form,
|
||||
badges=badges,
|
||||
users=User.query.order_by(User.is_admin.desc(), User.name.asc()).all(),
|
||||
push_ready=push_enabled(),
|
||||
vapid_public_key=current_app.config["VAPID_PUBLIC_KEY"],
|
||||
has_subscription=bool(subscriptions),
|
||||
@@ -58,6 +68,8 @@ def index():
|
||||
@bp.route("/badges/<int:badge_id>", methods=["POST"])
|
||||
@login_required
|
||||
def update_badge(badge_id: int):
|
||||
if not _require_admin():
|
||||
return redirect(url_for("settings.index"))
|
||||
badge = BadgeDefinition.query.get_or_404(badge_id)
|
||||
badge.threshold = max(1, request.form.get("threshold", type=int, default=badge.threshold))
|
||||
badge.bonus_points = max(0, request.form.get("bonus_points", type=int, default=badge.bonus_points))
|
||||
@@ -67,6 +79,83 @@ def update_badge(badge_id: int):
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
|
||||
@bp.route("/users", methods=["POST"])
|
||||
@login_required
|
||||
def create_user():
|
||||
if not _require_admin():
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
form = AdminUserForm(prefix="admin")
|
||||
if not form.validate_on_submit():
|
||||
for field_errors in form.errors.values():
|
||||
for error in field_errors:
|
||||
flash(error, "error")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
user = User(
|
||||
name=form.name.data.strip(),
|
||||
email=form.email.data.lower().strip(),
|
||||
is_admin=form.is_admin.data,
|
||||
)
|
||||
user.set_password(form.password.data)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
flash(f"Nutzer „{user.name}“ wurde angelegt.", "success")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
|
||||
@bp.route("/users/<int:user_id>/toggle-admin", methods=["POST"])
|
||||
@login_required
|
||||
def toggle_admin(user_id: int):
|
||||
if not _require_admin():
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
user = User.query.get_or_404(user_id)
|
||||
make_admin = request.form.get("make_admin") == "1"
|
||||
|
||||
if user.id == current_user.id and not make_admin:
|
||||
flash("Du kannst dir die Admin-Rechte nicht selbst entziehen.", "error")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
if not make_admin and User.query.filter_by(is_admin=True).count() <= 1:
|
||||
flash("Mindestens ein Admin muss erhalten bleiben.", "error")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
user.is_admin = make_admin
|
||||
db.session.commit()
|
||||
flash(f"Admin-Status für „{user.name}“ wurde aktualisiert.", "success")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
|
||||
@bp.route("/users/<int:user_id>/delete", methods=["POST"])
|
||||
@login_required
|
||||
def delete_user(user_id: int):
|
||||
if not _require_admin():
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
user = User.query.get_or_404(user_id)
|
||||
|
||||
if user.id == current_user.id:
|
||||
flash("Du kannst deinen aktuell eingeloggten Account nicht löschen.", "error")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
if user.is_admin and User.query.filter_by(is_admin=True).count() <= 1:
|
||||
flash("Der letzte Admin kann nicht gelöscht werden.", "error")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
TaskTemplate.query.filter_by(default_assigned_user_id=user.id).update({"default_assigned_user_id": None})
|
||||
TaskInstance.query.filter_by(assigned_user_id=user.id).update({"assigned_user_id": None})
|
||||
TaskInstance.query.filter_by(completed_by_user_id=user.id).update({"completed_by_user_id": None})
|
||||
MonthlyScoreSnapshot.query.filter_by(user_id=user.id).delete()
|
||||
NotificationLog.query.filter_by(user_id=user.id).delete()
|
||||
PushSubscription.query.filter_by(user_id=user.id).delete()
|
||||
db.session.delete(user)
|
||||
db.session.commit()
|
||||
|
||||
flash("Nutzer wurde entfernt.", "success")
|
||||
return redirect(url_for("settings.index"))
|
||||
|
||||
|
||||
@bp.route("/push/subscribe", methods=["POST"])
|
||||
@login_required
|
||||
@csrf.exempt
|
||||
|
||||
Reference in New Issue
Block a user