75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""
|
|
数据库连接配置
|
|
"""
|
|
import pymysql
|
|
from contextlib import contextmanager
|
|
from lib.logger import log_info, log_error, log_warning
|
|
|
|
# 数据库配置
|
|
DB_CONFIG = {
|
|
'host': 'haoslm2.xicp.net',
|
|
'port': 10216,
|
|
'user': 'root',
|
|
'password': 'DsideaL147258369',
|
|
'database': 'yltcharge',
|
|
'charset': 'utf8mb4',
|
|
'cursorclass': pymysql.cursors.DictCursor,
|
|
'autocommit': True,
|
|
'connect_timeout': 30, # 连接超时30秒
|
|
'read_timeout': 60, # 读取超时60秒
|
|
'write_timeout': 60 # 写入超时60秒
|
|
}
|
|
|
|
|
|
@contextmanager
|
|
def get_connection():
|
|
"""获取数据库连接（上下文管理器）"""
|
|
try:
|
|
conn = pymysql.connect(**DB_CONFIG)
|
|
yield conn
|
|
except pymysql.Error as e:
|
|
log_error(f'数据库连接失败: {e}', 'db', exc_info=True)
|
|
raise
|
|
finally:
|
|
try:
|
|
conn.close()
|
|
except:
|
|
pass
|
|
|
|
|
|
def execute_query(sql, params=None):
|
|
"""执行查询并返回结果"""
|
|
try:
|
|
with get_connection() as conn:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute(sql, params)
|
|
result = cursor.fetchall()
|
|
return result
|
|
except Exception as e:
|
|
log_error(f'查询失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True)
|
|
raise
|
|
|
|
|
|
def execute_update(sql, params=None):
|
|
"""执行更新操作"""
|
|
try:
|
|
with get_connection() as conn:
|
|
with conn.cursor() as cursor:
|
|
affected = cursor.execute(sql, params)
|
|
return cursor.rowcount
|
|
except Exception as e:
|
|
log_error(f'更新失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True)
|
|
raise
|
|
|
|
|
|
def execute_insert(sql, params=None):
|
|
"""执行插入操作并返回插入ID"""
|
|
try:
|
|
with get_connection() as conn:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute(sql, params)
|
|
last_id = cursor.lastrowid
|
|
return last_id
|
|
except Exception as e:
|
|
log_error(f'插入失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True)
|
|
raise |