153 lines
5.1 KiB
Python
153 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Datenbank-Migrationsskript für Guest-Requests, UserPermissions und Notifications
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import sqlite3
|
|
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
|
|
from config.settings import DATABASE_PATH
|
|
|
|
logger = get_logger("migrate")
|
|
|
|
def column_exists(cursor, table_name, column_name):
|
|
"""Prüft, ob eine Spalte in einer Tabelle existiert."""
|
|
cursor.execute(f"PRAGMA table_info({table_name})")
|
|
columns = [row[1] for row in cursor.fetchall()]
|
|
return column_name in columns
|
|
|
|
def get_database_path():
|
|
"""Ermittelt den Pfad zur Datenbankdatei."""
|
|
# Verwende den korrekten Datenbankpfad aus der Konfiguration
|
|
if os.path.exists(DATABASE_PATH):
|
|
return DATABASE_PATH
|
|
|
|
# Fallback für alternative Pfade mit korrektem Dateinamen
|
|
alternative_paths = [
|
|
os.path.join('database', 'myp.db'),
|
|
'myp.db',
|
|
'../database/myp.db',
|
|
'./database/myp.db'
|
|
]
|
|
|
|
for path in alternative_paths:
|
|
if os.path.exists(path):
|
|
return path
|
|
|
|
# Falls keine Datei gefunden wird, verwende den konfigurierten Pfad
|
|
return DATABASE_PATH
|
|
|
|
def migrate_guest_requests_table():
|
|
"""Migriert die guest_requests Tabelle für neue Spalten."""
|
|
db_path = get_database_path()
|
|
|
|
if not os.path.exists(db_path):
|
|
logger.warning(f"Datenbankdatei nicht gefunden: {db_path}")
|
|
return False
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# Prüfen ob otp_used_at Spalte bereits existiert
|
|
if not column_exists(cursor, 'guest_requests', 'otp_used_at'):
|
|
cursor.execute("""
|
|
ALTER TABLE guest_requests
|
|
ADD COLUMN otp_used_at DATETIME
|
|
""")
|
|
logger.info("Spalte 'otp_used_at' zur guest_requests Tabelle hinzugefügt")
|
|
else:
|
|
logger.info("Spalte 'otp_used_at' existiert bereits")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Migration der guest_requests Tabelle: {str(e)}")
|
|
if 'conn' in locals():
|
|
conn.rollback()
|
|
conn.close()
|
|
return False
|
|
|
|
def main():
|
|
"""Führt die Datenbank-Migration aus."""
|
|
try:
|
|
logger.info("Starte Datenbank-Migration...")
|
|
|
|
# Datenbank initialisieren (erstellt neue Tabellen)
|
|
init_db()
|
|
|
|
# Spezifische Spalten-Migrationen
|
|
logger.info("Führe spezifische Tabellen-Migrationen aus...")
|
|
migrate_guest_requests_table()
|
|
|
|
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() |