47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import mysql.connector
|
|
from mysql.connector import Error
|
|
|
|
from .config import MYSQL_CONFIG
|
|
|
|
|
|
class Database:
|
|
"""MySQL 连接池管理类"""
|
|
pool_config = {
|
|
"host": MYSQL_CONFIG.get("host", "localhost"),
|
|
"port": int(MYSQL_CONFIG.get("port", 3306)),
|
|
"user": MYSQL_CONFIG.get("user", "root"),
|
|
"password": MYSQL_CONFIG.get("password", ""),
|
|
"database": MYSQL_CONFIG.get("database", ""),
|
|
"charset": MYSQL_CONFIG.get("charset", "utf8mb4"),
|
|
"pool_name": "fastapi_pool",
|
|
"pool_size": 5,
|
|
"pool_reset_session": True
|
|
}
|
|
|
|
@classmethod
|
|
def get_connection(cls):
|
|
"""获取数据库连接"""
|
|
try:
|
|
# 从连接池获取连接
|
|
conn = mysql.connector.connect(**cls.pool_config)
|
|
if conn.is_connected():
|
|
return conn
|
|
except Error as e:
|
|
# 抛出数据库连接错误（会被全局异常处理器捕获）
|
|
raise Exception(f"MySQL 连接失败: {str(e)}") from e
|
|
|
|
@classmethod
|
|
def close_connection(cls, conn, cursor=None):
|
|
"""关闭连接和游标"""
|
|
try:
|
|
if cursor:
|
|
cursor.close()
|
|
if conn and conn.is_connected():
|
|
conn.close()
|
|
except Error as e:
|
|
raise Exception(f"MySQL 连接关闭失败: {str(e)}") from e
|
|
|
|
|
|
# 暴露数据库操作工具
|
|
db = Database()
|