{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/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"
}
}
{
"tool_response": {
"type": "update",
"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",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 11,
"newStart": 1,
"newLines": 10,
"lines": [
" \"\"\"Database Connection Pool\"\"\"",
"-import sys",
"+",
" from contextlib import contextmanager",
" ",
" import mysql.connector",
" from mysql.connector import pooling",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")",
" from config import Config",
" ",
" "
]
},
{
"oldStart": 16,
"oldLines": 7,
"newStart": 15,
"newLines": 7,
"lines": [
" ",
" @classmethod",
" def get_pool(cls):",
"- \"\"\"Erstellt oder gibt existierenden Connection Pool zurück\"\"\"",
"+ \"\"\"Erstellt oder gibt existierenden Connection Pool zurueck\"\"\"",
" if cls._pool is None:",
" cls._pool = pooling.MySQLConnectionPool(",
" pool_name=\"mcp_pool\","
]
},
{
"oldStart": 34,
"oldLines": 7,
"newStart": 33,
"newLines": 7,
"lines": [
" @contextmanager",
" def get_connection(cls, database: str):",
" \"\"\"",
"- Context Manager für DB Connection.",
"+ Context Manager fuer DB Connection.",
" ",
" Args:",
" database: Datenbankname (bereits validiert)"
]
},
{
"oldStart": 51,
"oldLines": 7,
"newStart": 50,
"newLines": 7,
"lines": [
" f\"SET SESSION max_statement_time = {Config.QUERY_TIMEOUT_SEC}\"",
" )",
" except Exception:",
"- pass # Falls Variable nicht unterstützt wird",
"+ pass # Falls Variable nicht unterstuetzt wird",
" cursor.close()",
" ",
" yield conn"
]
}
],
"originalFile": "\"\"\"Database Connection Pool\"\"\"\nimport sys\nfrom contextlib import contextmanager\n\nimport mysql.connector\nfrom mysql.connector import pooling\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")\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 zurück\"\"\"\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 für 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 unterstützt wird\n cursor.close()\n\n yield conn\n finally:\n conn.close()\n"
}
}