mysqlpulse

MySQLPulse 详细使用文档

一、框架概述

MySQLPulse 是基于 dbutils.pooled_db 实现的高性能 MySQL 操作工具，封装了数据库连接池、SQL 构建、缓存优化等功能，支持 6 大核心操作：

• 创建表（create）
• 单条插入（insert）
• 条件查询（select，带 LRU 缓存）
• 条件更新（update）
• 条件删除（delete）
• 原生 SQL 执行（sql）

核心特性：

• 连接池优化：自动根据 CPU 核心数配置连接池大小，支持高并发场景
• 类型安全：通过 typing 注解定义参数类型
• 日志系统：详细记录操作耗时、影响行数、错误信息
• 条件构建器：支持复杂条件（IN/NULL/自定义操作符）
• 查询缓存：对 select 方法结果进行缓存（默认缓存 1024 条）

二、安装与依赖

pip install mysqlpulse

三、初始化配置

3.1 构造函数参数

from MySQLPulse import MySQLPulse

db = MySQLPulse(
    dbName="your_database",       # 数据库名（必填，默认 "mydb"）
    host="127.0.0.1",             # 数据库地址（默认 "localhost"）
    user="root",                   # 用户名（默认 "root"）
    password="your_password",      # 密码（默认 "root"）
    charset="utf8mb4",            # 字符集（默认 "utf8mb4"，支持 emoji）
    poolSize=10                    # 连接池大小（默认 CPU 核心数 × 5）
)

3.2 日志配置

默认使用 logging 输出 INFO 级日志，格式：
[HH:MM:SS] [INFO] 操作状态 | 详细信息
可通过修改 logging.basicConfig 自定义日志行为：

import logging
logging.basicConfig(
    level=logging.DEBUG,          # 调整日志级别
    filename="db.log",            # 日志写入文件
    format="%(asctime)s - %(message)s"
)

四、核心方法详解

4.1 创建数据表（create(tableName, columns)）

参数：

• tableName：表名（字符串）
• columns：列定义字典（键=列名，值=SQL 数据类型，如 "id": "INT AUTO_INCREMENT PRIMARY KEY"）

示例：

# 创建用户表（含主键、唯一索引）
columns = {
    "user_id": "BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT",
    "username": "VARCHAR(50) UNIQUE NOT NULL",  # 唯一约束
    "email": "VARCHAR(100) NOT NULL",
    "create_time": "DATETIME DEFAULT CURRENT_TIMESTAMP"  # 默认值
}
success = db.create("users", columns)
# 输出：[HH:MM:SS] [INFO] 📝 ✅ 表创建成功: users

4.2 插入单条数据（insert(tableName, data)）

参数：

• tableName：表名
• data：键值对字典（键=列名，值=插入值，支持 None）

示例：

# 插入单条用户数据
data = {
    "username": "alice",
    "email": "[e-mail]",
    "age": 30  # 允许数值类型
}
rows = db.insert("users", data)
# 输出：[HH:MM:SS] [INFO] ⚡ ✅ 执行成功 | 影响行数: 1

4.3 条件查询（select(tableName, where=None, fields=None)）

参数：

• tableName：表名
• where：条件字典（支持多种语法，见下方）
• fields：查询字段列表（默认 *，支持字段别名）

条件构造语法：

条件类型	示例字典	生成 SQL
等值查询	{"username": "alice"}	WHERE username = %s
NULL 判断	{"email": None}	WHERE email IS NULL
范围查询	{"age": (">", 25)}	WHERE age > %s
IN 列表	{"user_id": [1, 2, 3]}	WHERE user_id IN (%s, %s, %s)
模糊查询	{"username": ("LIKE", "%alice%")}	WHERE username LIKE %s
多条件组合	{"age": (">", 25), "status": 1}	WHERE age > %s AND status = %s

示例：

# 查询年龄 20-30 岁且状态为启用的用户（带字段过滤）
where = {
    "age": (">=", 20),        # 大于等于 20
    "age__lt": ( "<", 30),   # 别名写法（避免键冲突，实际生成 "age < %s"）
    "status": 1              # 等值条件
}
fields = ["username", "email", "age"]  # 只查询指定字段
users = db.select("users", where=where, fields=fields)
# 输出：[HH:MM:SS] [INFO] 🔍 ✅ 查询完成 | 结果数: 5

缓存说明：

• 自动对 where 和 fields 进行哈希化缓存
• 缓存大小 maxsize=1024（可修改 _selectCached 装饰器参数）
• 清空缓存：db._selectCached.cache_clear()（谨慎使用）

4.4 条件更新（update(tableName, setData, where)）

参数：

• tableName：表名
• setData：更新的键值对（格式同 insert 的 data）
• where：更新条件（格式同 select 的 where，必须包含条件，避免全表更新）

示例：

# 将用户名为 "alice" 的年龄加 1，并标记为已验证
set_data = {
    "age": "age + 1",        # 支持 SQL 表达式（直接写入，需注意安全）
    "is_verified": 1         # 普通键值对
}
where = {
    "username": "alice"
}
rows = db.update("users", set_data, where)
# 输出：[HH:MM:SS] [INFO] ⚡ ✅ 执行成功 | 影响行数: 1

安全提示：

• 必须传入 where 条件，否则抛出异常（防止误操作）
• 表达式更新（如 age + 1）需手动拼接，建议使用参数化避免 SQL 注入

4.5 条件删除（delete(tableName, where=None)）

参数：

• tableName：表名
• where：删除条件（格式同 select 的 where，无默认条件时删除全表数据！）

示例：

# 删除 90 天前的日志数据（使用日期函数）
where = {
    "create_time": ("<", "2023-01-01 00:00:00")
}
rows = db.delete("logs", where)
# 输出：[HH:MM:SS] [INFO] ⚡ ✅ 执行成功 | 影响行数: 1000

# 危险操作：删除全表（必须显式传入 `where=None`）
rows = db.delete("users", where=None)  # 慎用！

警告：

• 无 where 条件时会删除全表数据，调用前请确认！
• 建议配合 select 先验证条件正确性

4.6 原生 SQL 执行（sql(sql, args=None)）

参数：

• sql：原生 SQL 语句（可包含 {table} 占位符，自动替换表名）
• args：参数列表（防止 SQL 注入，使用 %s 占位）

示例：

# 带表名占位符的 SQL
table = "users"
sql = "SELECT COUNT(*) AS total FROM {table} WHERE age > %s"
count = db.sql(sql, args=(25,))  # 自动替换 {table} 为 "users"
# 输出：[{"total": 15}]

# 执行带 JOIN 的复杂查询
sql = """
SELECT u.username, o.order_id
FROM users u
JOIN orders o ON u.user_id = o.user_id
WHERE o.create_time BETWEEN %s AND %s
"""
args = ("2023-01-01", "2023-12-31")
results = db.sql(sql, args)

返回值：

• SELECT 语句：返回字典列表（List[Dict]）
• 其他语句：返回影响行数（int）

五、高级功能

5.1 连接池配置

• mincached=2：最小空闲连接数（启动时创建）
• maxconnections=poolSize：最大连接数（默认 CPU 核心数 × 5）
• 自动回收连接，避免资源泄漏

5.2 错误处理

所有数据库操作错误（如连接失败、SQL 语法错误）均会抛出 MySQLdb.Error 异常，需在调用层捕获：

try:
    db.insert("non_exist_table", {})
except Error as e:
    print(f"数据库错误：{str(e)}")
    # 处理重试或回滚逻辑

5.3 性能优化建议

• 批量操作：
  原生 SQL 执行时使用 INSERT INTO ... VALUES (%s,%s), (%s,%s) 格式提升插入速度
• 索引优化：
  在 create 方法中定义索引（如 UNIQUE, INDEX）
• 缓存调优：
  根据查询频率调整 _selectCached(maxsize=2048) 缓存大小

六、完整类定义

class MySQLHelper:
    def __init__(self, dbName="mydb", host='localhost', user='root', password='root', charset='utf8mb4', poolSize=None):
        """初始化连接池，创建数据库（若不存在）"""
    
    def create(self, tableName: str, columns: Dict[str, str]) -> bool:
        """创建表，返回是否成功"""
    
    def insert(self, tableName: str, data: Dict) -> int:
        """插入单条数据，返回影响行数"""
    
    def select(self, tableName: str, where: Dict = None, fields: List[str] = None) -> List[Dict]:
        """带缓存的条件查询，返回结果列表"""
    
    def update(self, tableName: str, setData: Dict, where: Dict) -> int:
        """条件更新，返回影响行数"""
    
    def delete(self, tableName: str, where: Dict = None) -> int:
        """条件删除，返回影响行数"""
    
    def sql(self, sql: str, args: Tuple = None) -> Union[int, List[Dict]]:
        """执行原生 SQL，自动区分查询/非查询"""

七、版本说明

1.0.0（当前版本）

• 基础功能完备
• 支持复杂条件构建
• 连接池与缓存优化

后续计划

• 批量插入（insert_many）
• 事务支持（begin/commit/rollback）
• ORM 风格查询（链式调用）