58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
"""MySQL 数据库连接池模块"""
|
|
import mysql.connector
|
|
from mysql.connector import pooling
|
|
from contextlib import contextmanager
|
|
|
|
class Database:
|
|
_pool = None
|
|
|
|
@classmethod
|
|
def get_pool(cls):
|
|
if cls._pool is None:
|
|
cls._pool = pooling.MySQLConnectionPool(
|
|
pool_name="coolbot_pool",
|
|
pool_size=5,
|
|
host="127.0.0.1",
|
|
port=3306,
|
|
user="root",
|
|
password="Coolbot123",
|
|
database="coolbot_data",
|
|
charset="utf8mb4"
|
|
)
|
|
return cls._pool
|
|
|
|
@contextmanager
|
|
def get_connection(self):
|
|
conn = self.get_pool().get_connection()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
@contextmanager
|
|
def get_cursor(self, dictionary=True):
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor(dictionary=dictionary)
|
|
try:
|
|
yield cursor
|
|
conn.commit()
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
cursor.close()
|
|
|
|
db = Database()
|
|
|
|
def test_connection():
|
|
"""测试数据库连接"""
|
|
try:
|
|
with db.get_cursor() as cursor:
|
|
cursor.execute("SELECT 1 as test")
|
|
result = cursor.fetchone()
|
|
print(f"✅ 数据库连接成功: {result}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 数据库连接失败: {e}")
|
|
return False
|