Add initialize admin-account in migration

app/migrate.py

@@ -4,6 +4,9 @@ from pathlib import Path
 import logging
 from logging_config import setup_logging
 import os
+from contextlib import closing
+import getpass
+from auth import create_user
 
 APP_ENV = os.environ.get("APP_ENV", "dev")
 logger = setup_logging(APP_ENV)
@@ -17,7 +20,7 @@ BASE_DIR = Path(__file__).resolve().parents[1]
 DB_DIR = BASE_DIR / "data"
 DB_PATH = DB_DIR / "app.db"
 MIGRATIONS_DIR = BASE_DIR / "migrations"
-
+ADMIN_USERNAME = "admin"
 
 
 def get_connection() -> sqlite3.Connection:
@@ -74,11 +77,45 @@ def apply_migrations():
                 # hier nicht weiter machen
                 raise
         logger.info(f"Migrationen abgeschlossen. DB: {DB_PATH}")
-        #print(f"Migrationen abgeschlossen. DB: {DB_PATH}")
+
+        create_admin_user()
 
     finally:
         conn.close()
 
 
+def admin_exists() -> bool:
+    """Prüft, ob der Admin-User bereits existiert."""
+    with closing(get_connection()) as conn:
+        row = conn.execute(
+            "SELECT 1 FROM users WHERE username = ?",
+            (ADMIN_USERNAME,),
+        ).fetchone()
+    return row is not None
+
+
+def create_admin_user():
+    if admin_exists():
+        logger.info("Adminkonto existiert bereits! Kein initiales Konto angelegt.")
+        return
+    logger.info("Adminkonto wird angelegt ...")
+
+    pw1 = getpass.getpass("Passwort: ")
+    pw2 = getpass.getpass("Passwort wiederholen: ")
+
+    if pw1 != pw2:
+        logger.warning("Passwörter stimmen nicht überein! Abbruch.")
+        return
+
+    ok = create_user(ADMIN_USERNAME, pw1, role="admin")
+
+    if ok:
+        logger.info(f"Admin-Benutzer '{ADMIN_USERNAME}' wurde angelegt.")
+    else:
+        # Sollte eigentlich nicht passieren, weil wir vorher geprüft haben,
+        # aber falls z.B. Parallelzugriff o.Ä.
+        logger.info(f"Admin-Benutzer '{ADMIN_USERNAME}' konnte nicht angelegt werden.")
+
+
 if __name__ == "__main__":
     apply_migrations()