mysqlpulse 1.0.0
A high-performance MySQL database helper with connection pooling and query caching; 支持连接池和查询缓存的高性能MySQL数据库操作工具
Requires Python: >=3.7
Keywords: mysql, database, helper, connection pool, query caching, MySQL, 数据库, 助手, 连接池, 查询缓存


# MySQLPulse 详细使用文档


## 一、框架概述
`MySQLPulse` 是基于 `dbutils.pooled_db` 实现的高性能 MySQL 操作工具，封装了数据库连接池、SQL 构建、缓存优化等功能，支持 **6 大核心操作**：  
- 创建表（`create`）  
- 单条插入（`insert`）  
- 条件查询（`select`，带 LRU 缓存）  
- 条件更新（`update`）  
- 条件删除（`delete`）  
- 原生 SQL 执行（`sql`）  

### 核心特性：
1. **连接池优化**：自动根据 CPU 核心数配置连接池大小，支持高并发场景  
2. **类型安全**：通过 `typing` 注解定义参数类型  
3. **日志系统**：详细记录操作耗时、影响行数、错误信息  
4. **条件构建器**：支持复杂条件（`IN`/`NULL`/自定义操作符）  
5. **查询缓存**：对 `select` 方法结果进行缓存（默认缓存 1024 条）  


## 二、安装与依赖
```bash
pip install mysqlpulse
```

## 三、初始化配置
### 3.1 构造函数参数
```python
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` 自定义日志行为：  
```python
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"`）  

#### 示例：
```python
# 创建用户表（含主键、唯一索引）
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`）  

#### 示例：
```python
# 插入单条用户数据
data = {
    "username": "alice",
    "email": "alice@example.com",
    "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` |

#### 示例：
```python
# 查询年龄 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`，**必须包含条件**，避免全表更新）  

#### 示例：
```python
# 将用户名为 "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`，**无默认条件时删除全表数据！**）  

#### 示例：
```python
# 删除 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` 占位）  

#### 示例：
```python
# 带表名占位符的 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` 异常，需在调用层捕获：  
```python
try:
    db.insert("non_exist_table", {})
except Error as e:
    print(f"数据库错误：{str(e)}")
    # 处理重试或回滚逻辑
```


### 5.3 性能优化建议
1. **批量操作**：  
   原生 SQL 执行时使用 `INSERT INTO ... VALUES (%s,%s), (%s,%s)` 格式提升插入速度  
2. **索引优化**：  
   在 `create` 方法中定义索引（如 `UNIQUE`, `INDEX`）  
3. **缓存调优**：  
   根据查询频率调整 `_selectCached(maxsize=2048)` 缓存大小  


## 六、完整类定义
```python
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 风格查询（链式调用）