Nirodan
9db922703b
migrations.py - schema_migrations table tracks applied versions (version, description, applied_at) - MIGRATIONS list is append-only; each entry is (version, description, sql) - backup() dumps all user-data tables to a timestamped JSON file in backups/ before any schema changes so data can be recovered if something goes wrong - run_migrations() is idempotent: already-applied versions are skipped Integration - app.py calls _run_startup_migrations() at module load so every restart applies any pending migrations (no-op if schema is current) - setup_routes.py calls run_migrations() after the initial setup form is submitted so all tables exist before the user hits the main page for the first time - notes.py and admin.py: removed all per-request CREATE TABLE DDL; schema is now owned entirely by the migration system Docker - docker-compose.dev.yml: add backups-data volume so JSON backups survive container restarts and rebuilds - Dockerfile: pre-create /app/backend/logs and /app/backend/backups so the directories exist even before volumes are mounted Adding future schema changes - Append a new (version, description, sql) tuple to MIGRATIONS in migrations.py - The next restart will detect it as pending, back up first, then apply it Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
270 lines
9.1 KiB
Python
270 lines
9.1 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from werkzeug.security import generate_password_hash
|
|
from auth.token import verify_token
|
|
from util.db_pool import get_connection
|
|
from util.logger import logger
|
|
|
|
admin_bp = Blueprint("admin", __name__)
|
|
|
|
_VALID_ROLES = {"user", "admin"}
|
|
|
|
|
|
def _require_admin():
|
|
user = verify_token()
|
|
if not user:
|
|
return None, (jsonify({"message": "Nicht autorisiert"}), 401)
|
|
if user.get("role") != "admin":
|
|
logger.warning("🚫 Adminbereich verweigert (kein Admin)")
|
|
return None, (jsonify({"message": "Adminrechte erforderlich"}), 403)
|
|
return user, None
|
|
|
|
|
|
@admin_bp.route("/api/admin/users", methods=["GET"])
|
|
def list_users():
|
|
_, err = _require_admin()
|
|
if err:
|
|
return err
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT id, username, role FROM users ORDER BY username ASC")
|
|
users = cur.fetchall()
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
return jsonify(users)
|
|
except Exception as e:
|
|
logger.error(f"[Admin list_users] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/users", methods=["POST"])
|
|
def create_user():
|
|
admin, err = _require_admin()
|
|
if err:
|
|
return err
|
|
data = request.get_json(silent=True) or {}
|
|
username = data.get("username", "").strip()
|
|
password = data.get("password", "")
|
|
role = data.get("role", "user")
|
|
if not username or not password:
|
|
return jsonify({"message": "Username und Passwort erforderlich"}), 400
|
|
if role not in _VALID_ROLES:
|
|
return jsonify({"message": f"Ungültige Rolle. Erlaubt: {', '.join(_VALID_ROLES)}"}), 400
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT id FROM users WHERE username=%s", (username,))
|
|
if cur.fetchone():
|
|
cur.close()
|
|
return jsonify({"message": "Nutzer existiert bereits"}), 409
|
|
cur.execute(
|
|
"INSERT INTO users (username, password, role) VALUES (%s, %s, %s)",
|
|
(username, generate_password_hash(password), role)
|
|
)
|
|
conn.commit()
|
|
new_id = cur.lastrowid
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
logger.info(f"✅ User erstellt: {username} durch {admin['username']}")
|
|
return jsonify({"id": new_id, "username": username, "role": role}), 201
|
|
except Exception as e:
|
|
logger.error(f"[Admin create_user] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/users/<int:user_id>", methods=["PUT"])
|
|
def update_user(user_id):
|
|
admin, err = _require_admin()
|
|
if err:
|
|
return err
|
|
data = request.get_json(silent=True) or {}
|
|
role = data.get("role")
|
|
password = data.get("password")
|
|
if role is None and password is None:
|
|
return jsonify({"message": "Nichts zu aktualisieren"}), 400
|
|
if role is not None and role not in _VALID_ROLES:
|
|
return jsonify({"message": f"Ungültige Rolle. Erlaubt: {', '.join(_VALID_ROLES)}"}), 400
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor()
|
|
if role:
|
|
cur.execute("UPDATE users SET role=%s WHERE id=%s", (role, user_id))
|
|
if password:
|
|
cur.execute(
|
|
"UPDATE users SET password=%s WHERE id=%s",
|
|
(generate_password_hash(password), user_id)
|
|
)
|
|
conn.commit()
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
logger.info(f"✏️ User aktualisiert (id={user_id}) durch {admin['username']}")
|
|
return jsonify({"message": "Aktualisiert"}), 200
|
|
except Exception as e:
|
|
logger.error(f"[Admin update_user] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/users/<int:user_id>", methods=["DELETE"])
|
|
def delete_user(user_id):
|
|
admin, err = _require_admin()
|
|
if err:
|
|
return err
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT username FROM users WHERE id=%s", (user_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
cur.close()
|
|
return jsonify({"message": "Nicht gefunden"}), 404
|
|
username = row[0]
|
|
if username == admin["username"]:
|
|
cur.close()
|
|
return jsonify({"message": "Du kannst dich nicht selbst löschen"}), 400
|
|
cur.execute("DELETE FROM users WHERE id=%s", (user_id,))
|
|
conn.commit()
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
logger.info(f"🗑️ User gelöscht (id={user_id}) durch {admin['username']}")
|
|
return jsonify({"message": "Gelöscht"}), 200
|
|
except Exception as e:
|
|
logger.error(f"[Admin delete_user] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
# ---------- Websites (Admin CRUD) ----------
|
|
|
|
@admin_bp.route("/api/admin/websites", methods=["GET"])
|
|
def list_websites_admin():
|
|
_, err = _require_admin()
|
|
if err:
|
|
return err
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
return jsonify(rows)
|
|
except Exception as e:
|
|
logger.error(f"[Admin list_websites] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/websites", methods=["POST"])
|
|
def create_website():
|
|
_, err = _require_admin()
|
|
if err:
|
|
return err
|
|
data = request.get_json(silent=True) or {}
|
|
name = data.get("name", "").strip()
|
|
url = data.get("url", "").strip()
|
|
description = data.get("description", "").strip()
|
|
if not name or not url:
|
|
return jsonify({"message": "Name und URL erforderlich"}), 400
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO websites (name, url, description) VALUES (%s, %s, %s)",
|
|
(name, url, description),
|
|
)
|
|
conn.commit()
|
|
new_id = cur.lastrowid
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
return jsonify({"id": new_id, "name": name, "url": url, "description": description}), 201
|
|
except Exception as e:
|
|
logger.error(f"[Admin create_website] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/websites/<int:item_id>", methods=["PUT"])
|
|
def update_website(item_id):
|
|
_, err = _require_admin()
|
|
if err:
|
|
return err
|
|
data = request.get_json(silent=True) or {}
|
|
name = data.get("name", "").strip()
|
|
url = data.get("url", "").strip()
|
|
if not name or not url:
|
|
return jsonify({"message": "Name und URL erforderlich"}), 400
|
|
description = data.get("description", "").strip()
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"UPDATE websites SET name=%s, url=%s, description=%s WHERE id=%s",
|
|
(name, url, description, item_id),
|
|
)
|
|
conn.commit()
|
|
affected = cur.rowcount
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
if affected == 0:
|
|
return jsonify({"message": "Nicht gefunden"}), 404
|
|
return jsonify({"message": "Aktualisiert"}), 200
|
|
except Exception as e:
|
|
logger.error(f"[Admin update_website] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
@admin_bp.route("/api/admin/websites/<int:item_id>", methods=["DELETE"])
|
|
def delete_website(item_id):
|
|
_, err = _require_admin()
|
|
if err:
|
|
return err
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM websites WHERE id=%s", (item_id,))
|
|
conn.commit()
|
|
affected = cur.rowcount
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
if affected == 0:
|
|
return jsonify({"message": "Nicht gefunden"}), 404
|
|
return jsonify({"message": "Gelöscht"}), 200
|
|
except Exception as e:
|
|
logger.error(f"[Admin delete_website] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|
|
|
|
|
|
# ---------- Public (logged-in) endpoints ----------
|
|
|
|
@admin_bp.route("/api/websites", methods=["GET"])
|
|
def list_websites_public():
|
|
user = verify_token()
|
|
if not user:
|
|
return jsonify({"message": "Nicht autorisiert"}), 401
|
|
try:
|
|
conn = get_connection()
|
|
try:
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
return jsonify(rows)
|
|
except Exception as e:
|
|
logger.error(f"[Public list_websites] {e}")
|
|
return jsonify({"message": "Serverfehler"}), 500
|