Dashboard Finished

config/database.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 from typing import Any, Dict
 
 
@@ -10,7 +11,15 @@ class DatabaseConfig:
 
     def _load_config(self) -> Dict[str, Any]:
         if self.db_type.lower() == "sqlite":
-            return {"path": os.getenv("SQLITE_PATH", "databases/chatbot.db")}
+            # Standardpfad relativ zum Projektroot
+            default_path = Path(__file__).parent.parent / "databases" / "chatbot.db"
+            sqlite_path = os.getenv("SQLITE_PATH", str(default_path))
+
+            # Stelle sicher, dass wir einen absoluten Pfad haben
+            if not Path(sqlite_path).is_absolute():
+                sqlite_path = Path.cwd() / sqlite_path
+
+            return {"path": str(sqlite_path)}
         elif self.db_type.lower() == "mysql":
             return {
                 "host": os.getenv("MYSQL_HOST", "localhost"),

database/sqlite_db.py

@@ -1,5 +1,6 @@
 import logging
 import sqlite3
+from pathlib import Path
 from threading import local
 from typing import Any, Dict, Optional
 
@@ -13,14 +14,30 @@ class SQLiteDatabase(DatabaseInterface):
 
     def __init__(self, db_path: str):
         self.db_path = db_path
+        # Erstelle Verzeichnis falls nicht vorhanden
+        self._ensure_database_directory()
         # Thread-local storage für Verbindungen
         self._local = local()
 
+    def _ensure_database_directory(self) -> None:
+        """Stellt sicher, dass das Database-Verzeichnis existiert"""
+        db_dir = Path(self.db_path).parent
+        if not db_dir.exists():
+            try:
+                db_dir.mkdir(parents=True, exist_ok=True)
+                logger.info(f"Created database directory: {db_dir}")
+            except OSError as e:
+                logger.error(f"Failed to create database directory {db_dir}: {e}")
+                raise
+
     def _get_connection(self) -> sqlite3.Connection:
         """Holt oder erstellt eine thread-lokale Verbindung"""
         if not hasattr(self._local, "connection") or self._local.connection is None:
+            try:
+                # Absoluten Pfad verwenden für bessere Kompatibilität
+                abs_path = Path(self.db_path).resolve()
                 self._local.connection = sqlite3.connect(
-                self.db_path,
+                    str(abs_path),
                     check_same_thread=False,  # Erlaubt thread-übergreifende Nutzung
                     timeout=30.0,
                     detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
@@ -31,14 +48,16 @@ class SQLiteDatabase(DatabaseInterface):
                 self._local.connection.execute("PRAGMA synchronous=NORMAL")
                 self._local.connection.execute("PRAGMA cache_size=1000")
                 self._local.connection.execute("PRAGMA temp_store=MEMORY")
-            logger.debug("New SQLite connection created for thread")
-
+                logger.debug(f"New SQLite connection created for thread: {abs_path}")
+            except sqlite3.Error as e:
+                logger.error(f"Failed to connect to SQLite database at {abs_path}: {e}")
+                raise
         return self._local.connection
 
     def connect(self) -> None:
         """Initialisiert die thread-lokale Verbindung"""
         try:
-            conn = self._get_connection()
+            self._get_connection()  # Entfernt das unbenutzte 'conn'
             logger.info(f"Connected to SQLite database: {self.db_path}")
         except Exception as e:
             logger.error(f"Failed to connect to SQLite database: {e}")

services/user_service.py

@@ -12,7 +12,9 @@ class UserService:
     def __init__(self, database: DatabaseInterface):
         self.db = database
 
-    def create_user(self, username: str, email: str, password: str) -> tuple[bool, list[str]]:
+    def create_user(
+        self, username: str, email: str, password: str
+    ) -> tuple[bool, list[str]]:
         errors = self._validate_user_inputs(username, email, password)
         if errors:
             return False, errors
@@ -24,7 +26,9 @@ class UserService:
         password_hash = PasswordUtils.hash_password_simple(password)
         return self._attempt_user_creation(username, email, password_hash)
 
-    def _validate_user_inputs(self, username: str, email: str, password: str) -> list[str]:
+    def _validate_user_inputs(
+        self, username: str, email: str, password: str
+    ) -> list[str]:
         errors = []
         if not ValidationUtils.validate_username(username):
             errors.append("Invalid username format")
@@ -43,7 +47,9 @@ class UserService:
             errors.append("Username already taken")
         return errors
 
-    def _attempt_user_creation(self, username: str, email: str, password_hash: str) -> tuple[bool, list[str]]:
+    def _attempt_user_creation(
+        self, username: str, email: str, password_hash: str
+    ) -> tuple[bool, list[str]]:
         try:
             success = self.db.create_user(username, email.lower(), password_hash)
             if success: