{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/scripts\/pipeline\/db_core.py",
"content": "\"\"\"\nDatabase Core - Connection Management\n\nSingle Responsibility: Database connection lifecycle and query execution.\nAll other database modules (mixins) inherit from this base class.\n\"\"\"\n\nimport mysql.connector\nfrom mysql.connector import Error\n\nfrom config import DB_CONFIG\n\n\nclass DatabaseCore:\n \"\"\"Base database class with connection management.\n\n Provides:\n - Connection lifecycle (connect, disconnect)\n - Query execution (execute, commit)\n\n All database mixins expect these methods to be available via self.\n \"\"\"\n\n def __init__(self):\n \"\"\"Initialize database instance without connection.\"\"\"\n self.connection = None\n\n def connect(self) -> bool:\n \"\"\"Establish database connection.\n\n Returns:\n True if connection successful, False otherwise.\n \"\"\"\n try:\n self.connection = mysql.connector.connect(**DB_CONFIG)\n return True\n except Error as e:\n print(f\"Database connection error: {e}\")\n return False\n\n def disconnect(self):\n \"\"\"Close database connection if open.\"\"\"\n if self.connection and self.connection.is_connected():\n self.connection.close()\n\n def execute(self, query: str, params: tuple = None):\n \"\"\"Execute a query and return the cursor.\n\n Args:\n query: SQL query string\n params: Optional tuple of query parameters\n\n Returns:\n Cursor with query results (dictionary mode)\n \"\"\"\n cursor = self.connection.cursor(dictionary=True)\n cursor.execute(query, params or ())\n return cursor\n\n def commit(self):\n \"\"\"Commit the current transaction.\"\"\"\n self.connection.commit()\n\n def is_connected(self) -> bool:\n \"\"\"Check if database connection is active.\n\n Returns:\n True if connected, False otherwise.\n \"\"\"\n return self.connection is not None and self.connection.is_connected()\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/scripts\/pipeline\/db_core.py",
"content": "\"\"\"\nDatabase Core - Connection Management\n\nSingle Responsibility: Database connection lifecycle and query execution.\nAll other database modules (mixins) inherit from this base class.\n\"\"\"\n\nimport mysql.connector\nfrom mysql.connector import Error\n\nfrom config import DB_CONFIG\n\n\nclass DatabaseCore:\n \"\"\"Base database class with connection management.\n\n Provides:\n - Connection lifecycle (connect, disconnect)\n - Query execution (execute, commit)\n\n All database mixins expect these methods to be available via self.\n \"\"\"\n\n def __init__(self):\n \"\"\"Initialize database instance without connection.\"\"\"\n self.connection = None\n\n def connect(self) -> bool:\n \"\"\"Establish database connection.\n\n Returns:\n True if connection successful, False otherwise.\n \"\"\"\n try:\n self.connection = mysql.connector.connect(**DB_CONFIG)\n return True\n except Error as e:\n print(f\"Database connection error: {e}\")\n return False\n\n def disconnect(self):\n \"\"\"Close database connection if open.\"\"\"\n if self.connection and self.connection.is_connected():\n self.connection.close()\n\n def execute(self, query: str, params: tuple = None):\n \"\"\"Execute a query and return the cursor.\n\n Args:\n query: SQL query string\n params: Optional tuple of query parameters\n\n Returns:\n Cursor with query results (dictionary mode)\n \"\"\"\n cursor = self.connection.cursor(dictionary=True)\n cursor.execute(query, params or ())\n return cursor\n\n def commit(self):\n \"\"\"Commit the current transaction.\"\"\"\n self.connection.commit()\n\n def is_connected(self) -> bool:\n \"\"\"Check if database connection is active.\n\n Returns:\n True if connected, False otherwise.\n \"\"\"\n return self.connection is not None and self.connection.is_connected()\n",
"structuredPatch": [],
"originalFile": null
}
}