63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
# database.py
|
|
import mysql.connector
|
|
from mysql.connector import Error
|
|
from contextlib import contextmanager
|
|
import logging
|
|
|
|
from app import Config
|
|
|
|
|
|
class DatabaseConfig:
|
|
def __init__(self):
|
|
self.host = Config.MYSQL_HOST
|
|
self.database = Config.MYSQL_DATABASE
|
|
self.user = Config.MYSQL_USER
|
|
self.password = Config.MYSQL_PASSWORD
|
|
self.port = Config.MYSQL_PORT
|
|
|
|
class DatabaseManager:
|
|
def __init__(self):
|
|
self.config = DatabaseConfig()
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
@contextmanager
|
|
def get_connection(self):
|
|
connection = None
|
|
try:
|
|
connection = mysql.connector.connect(
|
|
host=self.config.host,
|
|
database=self.config.database,
|
|
user=self.config.user,
|
|
password=self.config.password,
|
|
port=self.config.port,
|
|
charset='utf8mb4',
|
|
collation='utf8mb4_unicode_ci',
|
|
autocommit=True
|
|
)
|
|
yield connection
|
|
except Error as e:
|
|
self.logger.error(f"数据库连接错误: {e}")
|
|
raise
|
|
finally:
|
|
if connection and connection.is_connected():
|
|
connection.close()
|
|
|
|
@contextmanager
|
|
def get_cursor(self, connection=None):
|
|
if connection:
|
|
cursor = connection.cursor(dictionary=True)
|
|
try:
|
|
yield cursor
|
|
finally:
|
|
cursor.close()
|
|
else:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor(dictionary=True)
|
|
try:
|
|
yield cursor
|
|
conn.commit()
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
cursor.close() |