87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Datenbank-Migrationsskript für Guest-Requests, UserPermissions und Notifications
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
# Pfad zur App hinzufügen
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from models import init_db, get_cached_session, GuestRequest, UserPermission, Notification, User
|
|
from utils.logging_config import get_logger
|
|
|
|
logger = get_logger("migrate")
|
|
|
|
def main():
|
|
"""Führt die Datenbank-Migration aus."""
|
|
try:
|
|
logger.info("Starte Datenbank-Migration...")
|
|
|
|
# Datenbank initialisieren (erstellt neue Tabellen)
|
|
init_db()
|
|
|
|
logger.info("Datenbank-Migration erfolgreich abgeschlossen")
|
|
|
|
# Testen, ob die neuen Tabellen funktionieren
|
|
test_new_tables()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Datenbank-Migration: {str(e)}")
|
|
sys.exit(1)
|
|
|
|
def test_new_tables():
|
|
"""Testet, ob die neuen Tabellen korrekt erstellt wurden."""
|
|
try:
|
|
with get_cached_session() as session:
|
|
# Test der GuestRequest-Tabelle
|
|
test_request = GuestRequest(
|
|
name="Test User",
|
|
email="test@example.com",
|
|
reason="Test migration",
|
|
duration_min=60
|
|
)
|
|
session.add(test_request)
|
|
session.flush()
|
|
|
|
# Test der UserPermission-Tabelle (mit Admin-User falls vorhanden)
|
|
admin_user = session.query(User).filter_by(role="admin").first()
|
|
if admin_user:
|
|
# Prüfen, ob bereits Permissions für diesen User existieren
|
|
existing_permission = session.query(UserPermission).filter_by(user_id=admin_user.id).first()
|
|
|
|
if not existing_permission:
|
|
permission = UserPermission(
|
|
user_id=admin_user.id,
|
|
can_start_jobs=True,
|
|
needs_approval=False,
|
|
can_approve_jobs=True
|
|
)
|
|
session.add(permission)
|
|
session.flush()
|
|
logger.info(f"UserPermission für Admin-User {admin_user.id} erstellt")
|
|
else:
|
|
logger.info(f"UserPermission für Admin-User {admin_user.id} existiert bereits")
|
|
|
|
# Test der Notification-Tabelle
|
|
notification = Notification(
|
|
user_id=admin_user.id,
|
|
type="test",
|
|
payload='{"message": "Test notification"}'
|
|
)
|
|
session.add(notification)
|
|
session.flush()
|
|
|
|
# Test-Daten wieder löschen
|
|
session.rollback()
|
|
|
|
logger.info("Alle neuen Tabellen wurden erfolgreich getestet")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Testen der neuen Tabellen: {str(e)}")
|
|
raise
|
|
|
|
if __name__ == "__main__":
|
|
main() |