{
"event": "PreToolUse",
"tool_name": "Read",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/validators\/execute_validator.py"
}
}
{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/infrastructure\/db_connection.py",
"content": "\"\"\"Database Connection Pool\"\"\"\n\nfrom contextlib import contextmanager\n\nimport mysql.connector\nfrom mysql.connector import pooling\n\nfrom config import Config\n\n\nclass DatabaseConnection:\n \"\"\"Singleton Connection Pool - SRP: Nur Verbindungsmanagement\"\"\"\n\n _pool = None\n\n @classmethod\n def get_pool(cls):\n \"\"\"Erstellt oder gibt existierenden Connection Pool zurueck\"\"\"\n if cls._pool is None:\n cls._pool = pooling.MySQLConnectionPool(\n pool_name=\"mcp_pool\",\n pool_size=5,\n host=Config.DB_HOST,\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\",\n connection_timeout=10,\n autocommit=True,\n )\n return cls._pool\n\n @classmethod\n @contextmanager\n def get_connection(cls, database: str):\n \"\"\"\n Context Manager fuer DB Connection.\n\n Args:\n database: Datenbankname (bereits validiert)\n \"\"\"\n conn = cls.get_pool().get_connection()\n try:\n # Datenbank wechseln (USE statement statt property)\n cursor = conn.cursor()\n cursor.execute(f\"USE {database}\")\n\n # Query Timeout setzen (MariaDB verwendet max_statement_time in Sekunden)\n try:\n cursor.execute(\n f\"SET SESSION max_statement_time = {Config.QUERY_TIMEOUT_SEC}\"\n )\n except Exception:\n pass # Falls Variable nicht unterstuetzt wird\n cursor.close()\n\n yield conn\n finally:\n conn.close()\n",
"numLines": 59,
"startLine": 1,
"totalLines": 59
}
}
}