{
"event": "PostToolUse",
"tool_name": "Read",
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/shared\/infrastructure\/simple_db_connection.py",
"content": "\"\"\"\nSimple DB Connection without Pooling.\n\nNOT suitable for:\n- High-Throughput (> 50 queries\/minute)\n- Connection Pooling Requirements\n- mcp_db Server\n\nFor those cases: Use dedicated implementation with mysql.connector.pooling\n\"\"\"\n\nfrom contextlib import contextmanager\nfrom typing import TYPE_CHECKING, Generator\n\nimport pymysql\nfrom pymysql.connections import Connection\n\nif TYPE_CHECKING:\n from shared.config_base import AppDatabaseConfig, LogDatabaseConfig\n\n\nclass SimpleDbConnection:\n \"\"\"\n Simple database connection without pooling.\n\n For servers with low query volume (< 50\/minute).\n \"\"\"\n\n @classmethod\n @contextmanager\n def get_connection(\n cls,\n config: \"AppDatabaseConfig\",\n database: str | None = None,\n autocommit: bool = False,\n ) -> Generator[Connection, None, None]:\n \"\"\"\n Context Manager for App DB Connection.\n\n WARNING: Do not add more parameters!\n For more logic: Create a new method.\n\n Args:\n config: Server configuration with DB credentials\n database: Optional - overrides config.DB_NAME\n autocommit: True for logging, False for transactions\n \"\"\"\n conn = None\n db_name = database or getattr(config, \"DB_NAME\", None)\n\n try:\n conn = pymysql.connect(\n host=config.DB_HOST,\n port=getattr(config, \"DB_PORT\", 3306),\n user=config.DB_USER,\n password=config.DB_PASSWORD,\n database=db_name,\n charset=\"utf8mb4\",\n cursorclass=pymysql.cursors.DictCursor,\n autocommit=autocommit,\n )\n yield conn\n if not autocommit:\n conn.commit()\n except Exception:\n if conn and not autocommit:\n conn.rollback()\n raise\n finally:\n if conn:\n conn.close()\n\n @classmethod\n @contextmanager\n def get_log_connection(\n cls,\n config: \"LogDatabaseConfig\",\n ) -> Generator[Connection, None, None]:\n \"\"\"Separate connection for logging (autocommit=True).\"\"\"\n conn = None\n try:\n conn = pymysql.connect(\n host=config.LOG_DB_HOST,\n user=config.LOG_DB_USER,\n password=config.LOG_DB_PASSWORD,\n database=config.LOG_DB_NAME,\n charset=\"utf8mb4\",\n cursorclass=pymysql.cursors.DictCursor,\n autocommit=True,\n )\n yield conn\n finally:\n if conn:\n conn.close()\n",
"numLines": 95,
"startLine": 1,
"totalLines": 95
}
}
}