{
"event": "PostToolUse",
"tool_name": "Read",
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/migration\/content-pipeline\/src\/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",
"numLines": 71,
"startLine": 1,
"totalLines": 71
}
}
}