{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_code\/db.py",
"content": "\"\"\"DB-Wrapper für mcp-code - nutzt Shared Infrastructure.\"\"\"\n\nfrom config import Config\nfrom shared.infrastructure import SimpleDbConnection\n\n\ndef execute_query(query: str, params: tuple = None, max_rows: int = 100) -> list:\n \"\"\"Fuehrt Query aus und gibt Ergebnisse zurueck.\"\"\"\n with SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(query, params or ())\n return cursor.fetchmany(max_rows)\n\n\ndef execute_single(query: str, params: tuple = None) -> dict | None:\n \"\"\"Fuehrt Query aus und gibt einzelnes Ergebnis zurueck.\"\"\"\n rows = execute_query(query, params, max_rows=1)\n return rows[0] if rows else None\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_code\/db.py",
"content": "\"\"\"DB-Wrapper für mcp-code - nutzt Shared Infrastructure.\"\"\"\n\nfrom config import Config\nfrom shared.infrastructure import SimpleDbConnection\n\n\ndef execute_query(query: str, params: tuple = None, max_rows: int = 100) -> list:\n \"\"\"Fuehrt Query aus und gibt Ergebnisse zurueck.\"\"\"\n with SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(query, params or ())\n return cursor.fetchmany(max_rows)\n\n\ndef execute_single(query: str, params: tuple = None) -> dict | None:\n \"\"\"Fuehrt Query aus und gibt einzelnes Ergebnis zurueck.\"\"\"\n rows = execute_query(query, params, max_rows=1)\n return rows[0] if rows else None\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 30,
"newStart": 1,
"newLines": 18,
"lines": [
"-\"\"\"Einfache DB-Connection für mcp-code.\"\"\"",
"-import mysql.connector",
"+\"\"\"DB-Wrapper für mcp-code - nutzt Shared Infrastructure.\"\"\"",
"+",
" from config import Config",
"+from shared.infrastructure import SimpleDbConnection",
" ",
" ",
"-def get_connection():",
"- \"\"\"Erstellt eine DB-Connection.\"\"\"",
"- return mysql.connector.connect(",
"- host=Config.DB_HOST,",
"- user=Config.DB_USER,",
"- password=Config.DB_PASSWORD,",
"- database=Config.DB_NAME,",
"- )",
"-",
"-",
" def execute_query(query: str, params: tuple = None, max_rows: int = 100) -> list:",
"- \"\"\"Führt Query aus und gibt Ergebnisse zurück.\"\"\"",
"- conn = get_connection()",
"- cursor = conn.cursor(dictionary=True, buffered=True)",
"- cursor.execute(query, params or ())",
"- rows = cursor.fetchmany(max_rows)",
"- cursor.close()",
"- conn.close()",
"- return rows",
"+ \"\"\"Fuehrt Query aus und gibt Ergebnisse zurueck.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
"+ with conn.cursor() as cursor:",
"+ cursor.execute(query, params or ())",
"+ return cursor.fetchmany(max_rows)",
" ",
" ",
" def execute_single(query: str, params: tuple = None) -> dict | None:",
"- \"\"\"Führt Query aus und gibt einzelnes Ergebnis zurück.\"\"\"",
"+ \"\"\"Fuehrt Query aus und gibt einzelnes Ergebnis zurueck.\"\"\"",
" rows = execute_query(query, params, max_rows=1)",
" return rows[0] if rows else None"
]
}
],
"originalFile": "\"\"\"Einfache DB-Connection für mcp-code.\"\"\"\nimport mysql.connector\nfrom config import Config\n\n\ndef get_connection():\n \"\"\"Erstellt eine DB-Connection.\"\"\"\n return mysql.connector.connect(\n host=Config.DB_HOST,\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n database=Config.DB_NAME,\n )\n\n\ndef execute_query(query: str, params: tuple = None, max_rows: int = 100) -> list:\n \"\"\"Führt Query aus und gibt Ergebnisse zurück.\"\"\"\n conn = get_connection()\n cursor = conn.cursor(dictionary=True, buffered=True)\n cursor.execute(query, params or ())\n rows = cursor.fetchmany(max_rows)\n cursor.close()\n conn.close()\n return rows\n\n\ndef execute_single(query: str, params: tuple = None) -> dict | None:\n \"\"\"Führt Query aus und gibt einzelnes Ergebnis zurück.\"\"\"\n rows = execute_query(query, params, max_rows=1)\n return rows[0] if rows else None\n"
}
}