You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
65 lines
1.8 KiB
65 lines
1.8 KiB
from dbutils.pooled_db import PooledDB
|
|
import pymysql
|
|
|
|
from ZhuQue.Config.config import Config
|
|
|
|
config = Config()
|
|
|
|
|
|
# 连接池配置信息
|
|
def get_pool_config():
|
|
return {
|
|
'creator': pymysql,
|
|
'maxconnections': 100, # 根据您的需要调整
|
|
'mincached': 10, # 初始化时的连接数
|
|
'maxcached': 10, # 池中最多闲置的连接数
|
|
'maxshared': 10, # 池中最多共享的连接数量
|
|
'blocking': True, # 无可用连接时是否等待
|
|
'host': config.get('Database', 'HOST'),
|
|
'port': int(config.get('Database', 'PORT')),
|
|
'user': config.get('Database', 'USER'),
|
|
'password': config.get('Database', 'PASSWORD'),
|
|
'database': config.get('Database', 'DATABASE'),
|
|
'charset': 'utf8',
|
|
'cursorclass': pymysql.cursors.DictCursor # 使用字典形式的游标
|
|
}
|
|
|
|
|
|
# 初始化数据库连接池
|
|
def init_db_pool():
|
|
config = get_pool_config()
|
|
return PooledDB(**config)
|
|
|
|
|
|
# 确保只在应用启动时初始化一次
|
|
db_pool = init_db_pool()
|
|
|
|
|
|
# 从连接池获取连接
|
|
def get_db_connection():
|
|
return db_pool.connection()
|
|
|
|
|
|
# 执行SQL查询并返回结果的函数
|
|
def execute_query(sql, params=(), one=False):
|
|
conn = get_db_connection()
|
|
try:
|
|
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
|
cursor.execute(sql, params)
|
|
if one:
|
|
return cursor.fetchone()
|
|
else:
|
|
return cursor.fetchall()
|
|
except Exception as err:
|
|
print(err)
|
|
return None
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
|
|
# 示例：使用 execute_query 函数
|
|
def get_user_by_username(username):
|
|
sql = "SELECT * FROM users WHERE username = %s"
|
|
return execute_query(sql, (username,), one=True)
|