# PyMysqlTools

**Repository Path**: uraurara/PyMysqlTools

## Basic Information

- **Project Name**: PyMysqlTools
- **Description**: PyMysqlTools 是一个通过避免编写SQL语句的方式来操作mysql的工具库
- **Primary Language**: Python
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2022-01-08
- **Last Updated**: 2026-08-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: Python, MySQL, PyMysqlTools

## README

# PyMysqlTools

> 一个基于 PyMySQL 与 DBUtils 的 MySQL 工具库，用预置函数替代手写 SQL，让数据库操作更简单。

[![PyPI version](https://badge.fury.io/py/PyMysqlTools.svg)](https://pypi.org/project/PyMysqlTools/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Python](https://img.shields.io/badge/python-3.8%2B-blue.svg)](https://www.python.org/)

## 简介

PyMysqlTools 是一个预置了很多常用函数替代SQL语句来操作MySQL的工具库，旨在简化Python开发者与MySQL数据库的交互操作。

### 设计思路

项目采用职责分离的三层架构，让 SQL 的构造、执行与结果处理彼此解耦：

- **生成层**（`SqlGenerator` / `ClauseGenerator`）：把表名、字段、条件等 Python 数据结构拼装成合法的 SQL 字符串
- **执行层**（`SqlActuator`）：基于 PyMySQL 的 cursor 执行 SQL，自动处理事务提交（DML）与结果获取（DQL）
- **表示层**（`ResultSet`）：将数据库返回的元组统一封装为可迭代的 `dict` 或 `list` 结构，屏蔽底层差异

这种分层使得每一层都可独立替换或扩展，例如未来可平滑切换到其他数据库驱动而无需改动业务代码。

### 适用场景

- Web 后端、爬虫、数据脚本等需要频繁操作 MySQL 的 Python 项目
- 希望避免手写 SQL 的小型应用或快速原型开发
- 需要连接池支持的多线程并发场景
- 对结果集结构有定制需求（`dict` / `list` 互转）的场合

### 特性

- **零 SQL 编程**：通过 `insert_one`、`find_all`、`update_by_id` 等方法完成常见 CRUD
- **双连接模式**：支持普通直连与线程池连接（`PersistentDB` / `PooledDB`）
- **灵活的结果集**：查询结果可在 `dict` 与 `list` 两种结构间切换，支持全局配置或单次覆盖
- **表管理能力**：建表、清空、查看表结构、主键、自增值、估算行数等开箱即用
- **数据迁移**：一条命令把一张表的数据迁移到另一张表
- **可迭代结果集**：`ResultSet` 支持 `for ... in` 直接遍历，并自带 `get` / `limit` / `next` 等便捷方法

### 注意事项

- 主要面向中小型项目与教学场景；对复杂查询（多表 JOIN、子查询、窗口函数等）可通过 [`execute`](#原生-sql-执行) 方法执行原生 SQL
- DML 语句的值采用 `%s` 参数化占位符防注入；但 WHERE 条件中 dict 形式的值仍为字符串拼接，使用时请对输入做校验

## 环境要求

- Python ≥ 3.8.10
- MySQL ≥ 5.6
- 依赖：`PyMySQL~=1.0.2`、`DBUtils~=3.0.2`

## 安装

```bash
pip install PyMysqlTools
```

## 快速开始

### 1. 建立连接

```python
import PyMysqlTools

# 普通直连
mysql = PyMysqlTools.connect(
    database='db_test',
    username='root',
    password='123456'
)
print(mysql)  # <PyMysqlTools.main.Connect object>
```

### 2. 增删改查

```python
# 插入单条记录
mysql.insert_one('tb_test', {'username': 'abc', 'password': 'abc123'})

# 批量插入（支持 list[dict] 形式）
mysql.batch_insert('tb_test', [
    {'username': 'user1', 'password': 'pass1'},
    {'username': 'user2', 'password': 'pass2'},
])

# 根据 id 删除
mysql.delete_by_id('tb_test', 2)

# 按条件删除
mysql.delete_by('tb_test', condition={'status': -1})

# 根据 id 更新
mysql.update_by_id('tb_test', {'username': 'abc', 'password': 'abc123'}, 3)

# 按条件更新
mysql.update_by('tb_test', {'username': 'abc'}, condition={'status': 0})

# 查询全表并遍历
for row in mysql.find_all('tb_test'):
    print(row)

# 按条件查询指定字段
result = mysql.find_by('tb_test', fields=['id', 'username'], condition={'status': 1})

# 按 id 查询
result = mysql.find_by_id('tb_test', 3)
```

### 3. 表结构操作

```python
# 查看表结构
mysql.show_table_desc('tb_test')

# 查看表字段
mysql.show_table_fields('tb_test')

# 查询表记录条数（精确 / 估算）
mysql.show_table_size('tb_test')
mysql.show_table_vague_size('tb_test')

# 查看主键字段
mysql.show_table_primary_field('tb_test')

# 创建表（schema 支持 list 或 dict 形式）
schema = {
    'id': 'INT PRIMARY KEY AUTO_INCREMENT',
    'name': 'VARCHAR(255) NOT NULL',
    'age': 'INT DEFAULT 0',
}
mysql.create_table('tb_new', schema)

# 清空表数据
mysql.truncate_table('tb_test')
```

### 4. 结果集类型切换

默认结果集为 `dict` 结构：

```python
for row in mysql.find_all('tb_user'):
    print(row)

# {'id': 1, 'name': '张三', 'age': 14, 'phone': '15500000000'}
# {'id': 2, 'name': '李四', 'age': 18, 'phone': '15500000001'}
```

修改全局默认类型为 `list`：

```python
from PyMysqlTools import settings

settings.DEFAULT_RESULT_SET_TYPE = list
for row in mysql.find_all('tb_user'):
    print(row)

# [1, '张三', 14, '15500000000']
# [2, '李四', 18, '15500000001']
```

或仅本次查询使用 `list` 结构：

```python
for row in mysql.find_all('tb_user', list):
    print(row)
```

## 连接池

`PyMysqlTools` 内置两种连接池模式，通过 `ConnectType` 枚举选择：

```python
import PyMysqlTools
from PyMysqlTools import ConnectType

mysql = PyMysqlTools.connect_pool(
    ConnectType.persistent_db,
    {
        "username": "root",
        "password": "123456",
        "database": "db_test",
    }
)
print(mysql)  # <PyMysqlTools.main.ConnectPool object>
```

两种模式的区别：

- `persistent_db`：持久连接池，每个线程独占一条连接
- `pooled_db`：共享连接池，多线程复用一组连接

### 连接池参数自定义

两种连接池的默认参数定义在 `settings` 中：

```python
# 默认的 persistent_db 连接池参数
DEFAULT_PERSISTENT_DB_POOL_ARGS = {
    'max_usage': None,
    'set_session': None,
    'failures': None,
    'ping': 1,
    'closeable': False,
    'thread_local': None,
}

# 默认的 pooled_db 连接池参数
DEFAULT_POOLED_DB_POOL_ARGS = {
    'min_cached': 0,
    'max_cached': 10,
    'max_shared': 0,
    'max_connections': 0,
    'blocking': False,
    'max_usage': None,
    'set_session': None,
    'reset': True,
    'failures': None,
    'ping': 1,
}
```

通过修改 `settings` 中的对应字典来调整参数。**注意：必须在创建连接前修改才会生效**。

```python
from PyMysqlTools import settings

settings.DEFAULT_PERSISTENT_DB_POOL_ARGS['max_usage'] = 8
mysql = PyMysqlTools.connect_pool(
    ConnectType.persistent_db,
    {
        "username": "root",
        "password": "123456",
        "database": "db_test",
    }
)
```

## API 速览

### 连接相关

| 方法 | 说明 |
| --- | --- |
| `connect(database, username, password, ...)` | 建立普通直连 |
| `connect_pool(connect_type, connect_args, **pool_args)` | 建立连接池连接 |
| `close()` | 关闭当前连接 |
| `reconnect()` | 重新建立连接 |
| `ping()` | 检测 MySQL 服务连接状态 |

### 数据操作（CRUD）

| 方法 | 说明 |
| --- | --- |
| `insert_one(tb_name, data)` | 插入单条记录 |
| `batch_insert(tb_name, data)` | 批量插入记录 |
| `update_insert(tb_name, data)` | 存在则更新，不存在则插入 |
| `update_by(tb_name, data, condition)` | 按条件更新 |
| `update_by_id(tb_name, data, id_)` | 按 id 更新 |
| `delete_by(tb_name, condition)` | 按条件删除 |
| `delete_by_id(tb_name, id_)` | 按 id 删除 |
| `find_by(tb_name, fields, condition, type_)` | 按条件查询 |
| `find_by_id(tb_name, id_, fields, type_)` | 按 id 查询 |
| `find_one(tb_name, fields, condition, type_)` | 查询单条记录 |
| `find_all(tb_name, type_)` | 查询全表记录 |

### 表与元信息

| 方法 | 说明 |
| --- | --- |
| `show_table_fields(tb_name, type_)` | 查看表字段 |
| `show_table_desc(tb_name, type_)` | 查看表结构 |
| `show_table_size(tb_name, type_)` | 查询表记录条数（精确） |
| `show_table_vague_size(tb_name, type_)` | 估算表记录条数（快速） |
| `show_auto_increment(tb_name, type_)` | 查看表的自增值 |
| `show_table_primary_field(tb_name, type_)` | 查询主键字段 |
| `show_databases(type_)` | 查看所有数据库 |
| `show_tables(type_)` | 查看所有数据表 |
| `is_exist_database(db_name)` | 判断数据库是否存在 |
| `is_exist_table(tb_name)` | 判断数据表是否存在 |

### 表管理

| 方法 | 说明 |
| --- | --- |
| `create_table(tb_name, schema)` | 创建数据表 |
| `create_table_not_exists(tb_name, schema)` | 表不存在时创建 |
| `truncate_table(tb_name)` | 清空表数据 |
| `delete_table(tb_name)` | 删除表所有记录 |
| `analyze_table(tb_name, type_)` | 分析并统计表信息 |
| `set_auto_increment(tb_name, auto_increment)` | 设置表自增值 |
| `migration_table(for_tb_name, to_tb_name)` | 把一张表数据迁移到另一张表 |

### `ResultSet` 结果集

| 方法 | 说明 |
| --- | --- |
| `all()` | 返回结果集列表 |
| `get(index)` | 获取指定索引的记录 |
| `get_key(key)` | 获取单条记录中指定字段的值 |
| `limit(num)` | 截取前 n 条结果 |
| `next()` | 获取下一条结果 |
| `for row in result_set` | 直接迭代遍历 |

### 原生 SQL 执行

对于多表 JOIN、子查询、窗口函数等复杂场景，可使用原生 SQL 执行接口：

| 方法 | 说明 |
| --- | --- |
| `execute(sql, args)` | 自动判断 DQL/DML，返回结果集 tuple 或受影响行数 |
| `execute_dml(sql, args)` | 明确执行 DML（INSERT/UPDATE/DELETE），自动提交，返回受影响行数 |
| `execute_dql(sql, args, type_)` | 明确执行 DQL（SELECT/SHOW），返回 `ResultSet` |

```python
# 自动判断：SELECT 开头视为 DQL，返回 tuple
rows = mysql.execute('SELECT a.id, b.name FROM tb_a a JOIN tb_b b ON a.id = b.id')

# 参数化查询（推荐，防注入）
rows = mysql.execute('SELECT * FROM tb_test WHERE id > %s', [10])

# 明确 DML，自动提交事务
affected = mysql.execute_dml('UPDATE tb_test SET status = %s WHERE id = %s', [1, 5])

# 明确 DQL，返回 ResultSet（支持 type_ 切换结果集结构）
result = mysql.execute_dql('SELECT * FROM tb_test WHERE id > %s', [10], list)
for row in result:
    print(row)
```

## 架构概览

```
用户代码
  └─ Connect / ConnectPool            (main.py)
       ├─ SqlGenerator                 ┐
       ├─ ClauseGenerator               ┤  拼装 SQL 字符串
       ├─ SqlActuator      (actuator.py)┤  执行 SQL，返回 tuple
       └─ ResultSet        (result_set.py) 包装 tuple 为 list/dict
```

- [`main.py`](PyMysqlTools/main.py)：连接与业务方法的入口，承载全部对外 API
- [`generator.py`](PyMysqlTools/generator.py)：负责生成 SQL 语句与 WHERE / SET / LIMIT 等子句
- [`actuator.py`](PyMysqlTools/actuator.py)：封装 cursor，区分 DML（自动 commit）与 DQL 查询
- [`result_set.py`](PyMysqlTools/result_set.py)：把查询结果统一封装为可迭代的 `ResultSet`
- [`settings.py`](PyMysqlTools/settings.py)：默认结果集类型与连接池参数
- [`exceptions.py`](PyMysqlTools/exceptions.py)：`TypeMismatchError`、`ParameterError` 等自定义异常

## 关于

如果您在使用过程中遇到了意料之外的结果，欢迎[提交 Issue](https://gitee.com/uraurara/PyMysqlTools/issues/new?issue%5Bassignee_id%5D=0&issue%5Bmilestone_id%5D=0)帮助我们改进此项目。

## 致谢

本项目在开发中使用了以下优秀的 Python 库：

- [PyMySQL](https://github.com/PyMySQL/PyMySQL) - MySQL 数据库连接驱动
- [DBUtils](https://github.com/WebwareForPython/DBUtils) - 数据库连接池管理

## License

本项目基于 [MIT License](LICENSE) 开源。