f3fbd0377a
- 实现微信登录、手机号绑定、token验证和登出功能 - 添加打卡系统，支持正常、迟到、请假等状态记录 - 集成课程投票功能，支持多课程选择和评分评论 - 开发反馈系统，支持图文反馈提交和管理员处理 - 构建首页页面模块，动态展示学习资源和个人信息 - 设计计划管理系统，支持周/月/年计划制定和跟踪 - 实现小组计划统计和历史记录查询功能 - 完善用户认证会话管理和数据库交互逻辑
98 lines
2.5 KiB
Python
98 lines
2.5 KiB
Python
"""Database access with a PyMySQL connection pool.
|
|
|
|
Each fetch/execute borrows a pooled connection and returns it on close().
|
|
Pool settings are read from environment (DB_POOL_*), so nothing here depends
|
|
on config.py and the pool can be shared safely across threads via DBUtils.
|
|
"""
|
|
import os
|
|
import logging
|
|
import threading
|
|
|
|
import pymysql
|
|
from dbutils.pooled_db import PooledDB
|
|
|
|
from . import config
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _pool_args(database):
|
|
return dict(
|
|
mincached=int(os.environ.get("DB_POOL_MIN_CACHED", "1")),
|
|
maxcached=int(os.environ.get("DB_POOL_MAX_CACHED", "5")),
|
|
maxconnections=int(os.environ.get("DB_POOL_MAX_CONNECTIONS", "10")),
|
|
blocking=True,
|
|
maxusage=int(os.environ.get("DB_POOL_MAX_USAGE", "100")),
|
|
ping=1,
|
|
host=config.MYSQL_HOST,
|
|
port=config.MYSQL_PORT,
|
|
user=config.MYSQL_USER,
|
|
password=config.MYSQL_PASSWORD,
|
|
database=database,
|
|
charset="utf8mb4",
|
|
connect_timeout=5,
|
|
read_timeout=5,
|
|
)
|
|
|
|
|
|
_pools = {}
|
|
_pools_lock = threading.Lock()
|
|
|
|
|
|
def _get_pool(database):
|
|
key = database or config.MYSQL_DB
|
|
with _pools_lock:
|
|
pool = _pools.get(key)
|
|
if pool is None:
|
|
pool = PooledDB(creator=pymysql, **_pool_args(key))
|
|
_pools[key] = pool
|
|
return pool
|
|
|
|
|
|
def _connect(db=None):
|
|
"""Borrow a pooled connection (close() returns it to the pool)."""
|
|
return _get_pool(db).connection()
|
|
|
|
|
|
def fetch_one(sql, args=None, db=None):
|
|
conn = _connect(db)
|
|
try:
|
|
with conn.cursor(pymysql.cursors.DictCursor) as cur:
|
|
cur.execute(sql, args or ())
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def fetch_all(sql, args=None, db=None):
|
|
conn = _connect(db)
|
|
try:
|
|
with conn.cursor(pymysql.cursors.DictCursor) as cur:
|
|
cur.execute(sql, args or ())
|
|
return cur.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def execute(sql, args=None, db=None):
|
|
conn = _connect(db)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, args or ())
|
|
lastrowid = cur.lastrowid
|
|
conn.commit()
|
|
return lastrowid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def close_all():
|
|
"""Close every pooled connection (used by tests / reload hooks)."""
|
|
with _pools_lock:
|
|
for key in list(_pools):
|
|
try:
|
|
_pools.pop(key).close()
|
|
except Exception:
|
|
logger.exception("关闭数据库连接池失败: %s", key)
|