35连接关系型数据库

关系型数据库，以MySQL为例。

Python 连接 MySQL 数据库是后端开发、数据分析、自动化脚本等场景中的常见需求。

## 核心库

Python 连接 MySQL 需依赖第三方库，主流选择有两个，功能类似但实现方式不同，需根据场景选择：

| 库名称                 | 特点                                                         | 适用场景                 |
| ---------------------- | ------------------------------------------------------------ | ------------------------ |
| mysql-connector-python | MySQL 官方库，完全兼容 MySQL 协议，更新及时，但性能略逊于第三方库 | 对官方兼容性要求高的场景 |
| PyMySQL                | 第三方库，纯 Python 实现，性能优异，社区活跃，支持 Python 3.6+，使用广泛 | 大多数开发场景（推荐）   |

## 安装

```cmd
pip install pymysql
```

## 概述

pymysql 是 Python 操作 MySQL 数据库的主流第三方库，其核心功能通过 **Connection（连接对象）** 和 **Cursor（游标对象）** 实现。

## Connection 对象

Connection 对象代表与 MySQL 服务器的连接，负责管理连接状态、事务控制等。通过 pymysql.connect() 创建，常用方法如下：

**1. pymysql.connect()**

作用：建立与 MySQL 服务器的连接，返回 Connection 对象。

常用参数：
| 参数            | 说明                             | 默认值             |
| --------------- | -------------------------------- | ------------------ |
| host            | 数据库主机地址                   | localhost          |
| port            | 端口号                           | 3306               |
| user            | 登录用户名（必填）               | -                  |
| password        | 登录密码（必填）                 | -                  |
| database        | 要连接的数据库名                 | None（需后续指定） |
| charset         | 字符集（支持中文需设为 utf8mb4） | 'utf8'             |
| autocommit      | 是否自动提交事务（True/False）   | False              |
| connect_timeout | 连接超时时间（秒）               | 10                 |

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    password="your_password",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)
```

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)
print(type(conn))
print(conn)
```

输出结果：
```txt
<class 'pymysql.connections.Connection'>
<pymysql.connections.Connection object at 0x0000021E4022D400>
```

**2. conn.cursor()**

游标（Cursor） 是一种数据库查询机制，它允许你从一组数据记录中每次逐行地读取一条记录，并对该行数据进行处理。

作用：生成 Cursor 对象（用于执行 SQL 语句）

参数：cursorclass：游标类型，默认返回元组，可选 pymysql.cursors.DictCursor（返回字典，键为列名）。

```python
# 创建默认游标（返回元组）
cursor = conn.cursor()

# 创建字典游标（返回字典，更直观）
dict_cursor = conn.cursor(pymysql.cursors.DictCursor)
```

**3. conn.close()**

作用：释放与数据库的连接，避免资源泄露。

注意：使用上下文管理器（with 语句）时会自动调用，无需手动关闭。

```python
conn.close()  # 手动关闭连接
```

**4. 其他常用方法**

| 方法                      | 作用                                     |
| ------------------------- | ---------------------------------------- |
| conn.select_db(db)        | 切换当前数据库（如从 test 切换到 test2） |
| conn.ping(reconnect=True) | 检查连接是否有效，断开时自动重连         |

## 事务控制方法

事务（Transaction） 是一个作为单个逻辑工作单元执行的一系列数据库操作（如插入、更新、删除等）的集合。

事务的四个核心特性（ACID），事务必须满足ACID特性：

| 特性                   | 描述                                                         | 在银行转账例子中的体现                                       |
| ---------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| **A**tomicity 原子性   | 事务中的所有操作作为一个不可分割的单元，要么全部成功，要么全部失败。 | 扣款和加款要么都发生，要么都不发生。                         |
| **C**onsistency 一致性 | 事务必须使数据库从一个 一致性状态变换到另一个 一致性状态，数据库的完整性约束不会被破坏。 | 转账前后，两个账户的总金额保持不变（A+B = (A-100)+(B+100)）。 |
| **I**solation 隔离性   | 一个事务的执行不能被其他事务干扰，并发执行的各个事务之间不能互相干扰。 | 在转账完成前，其他事务看不到中间的临时状态（例如，只看到A少了100，但B还没多100的状态）。 |
| **D**urability 持久性  | 一旦事务被提交，它对数据库中数据的改变就是永久性的，即使系统发生故障也不会丢失。 | 转账成功后，即使数据库服务器立即断电，重启后账户余额也已经是转账后的状态。 |

事务相关方法用于保证多步 SQL 操作的原子性（要么全成功，要么全失败）：

| 方法              | 作用                                         |
| ----------------- | -------------------------------------------- |
| conn.commit()     | 提交事务（使所有未生效的 SQL 操作生效）      |
| conn.rollback()   | 回滚事务（撤销所有未提交的操作）             |
| conn.autocommit() | 设置自动提交模式（True 则每条 SQL 自动提交） |

例如：
```python
try:
    # 执行插入操作
    cursor.execute("INSERT INTO users (name) VALUES (%s)", ("Test",))
    # 提交事务（生效）
    conn.commit()
except Exception as e:
    # 出错时回滚
    conn.rollback()
    print(f"事务失败：{e}")
```

## Cursor 对象

Cursor 对象是执行 SQL 语句的核心，负责发送 SQL 到服务器、获取结果等。常用方法如下：

**1. cursor.execute(sql, args=None)：执行单条 SQL**

作用：执行一条 SQL 语句（查询、插入、更新、删除等）。

参数：

- sql：SQL 语句字符串（需用 %s 作为参数占位符，而非 ? 或 %）。
- args：参数列表 / 字典（用于替换 %s，避免 SQL 注入）。

返回值：受影响的行数（查询时为匹配的总行数，插入 / 更新 / 删除时为修改的行数）。

```python
# 插入单条数据（参数为元组）
cursor.execute(
    "INSERT INTO users (name, age) VALUES (%s, %s)",
    ("Alice", 25)  # args 为元组，顺序对应 %s
)

# 查询数据（参数为字典）
cursor.execute(
    "SELECT * FROM users WHERE name = %(name)s",
    {"name": "Alice"}  # 字典参数，键对应 %(name)s
)
```

**2. cursor.executemany(sql, args_list)：批量执行 SQL**

作用：批量执行同一条 SQL 语句（如批量插入 / 更新），效率远高于多次调用 execute。

参数：

- sql：SQL 语句（同 execute）。
- args_list：参数列表（每个元素为一个 SQL 的参数，如 [(v1, v2), (v3, v4)]）。

返回值：所有语句受影响的总行数。

```python
# 批量插入 3 条用户数据
users = [
    ("Bob", 30, "bob@example.com"),
    ("Charlie", 35, "charlie@example.com"),
    ("David", 28, "david@example.com")
]
cursor.executemany(
    "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)",
    users  # 参数列表
)
conn.commit()  # 批量操作需提交事务
```

**3. 结果获取方法（查询专用）**

执行 SELECT 等查询语句后，需通过以下方法获取结果：

| 方法                   | 作用                            | 返回值                                  |
| ---------------------- | ------------------------------- | --------------------------------------- |
| cursor.fetchone()      | 获取结果集中的下一条数据        | 单条数据（元组或字典，无数据时为 None） |
| cursor.fetchall()      | 获取结果集中剩余的所有数据      | 列表（每个元素为一条数据）              |
| cursor.fetchmany(size) | 获取指定数量的结果（默认 1 条） | 列表（最多 size 条数据）                |

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    print(cursor.fetchone())
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：
```txt
(1, 'Alice', 26, 'alice@example.com')
```

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    for row in cursor.fetchall():
        print(row)
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：
```txt
(1, 'Alice', 26, 'alice@example.com')
(2, 'Bob', 30, 'bob@example.com')
```

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    for row in cursor.fetchmany(3):
        print(row)
        print("========")
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：
```txt
(1, 'Alice', 26, 'alice@example.com')
========
(2, 'Bob', 30, 'bob@example.com')
========
```

**4. 游标属性（结果信息）**

执行 SQL 后，可通过游标属性获取结果元数据：

| 属性               | 作用                                          | 示例值                        |
| ------------------ | --------------------------------------------- | ----------------------------- |
| cursor.rowcount    | 最后一条 SQL 影响的行数（查询时为匹配总行数） | 3（3 条数据被修改 / 查询）    |
| cursor.description | 查询结果的列信息（字段名、类型等）            | (('name', ...), ('age', ...)) |
| cursor.rownumber   | 当前游标在结果集中的位置（从 0 开始）         | 1（当前指向第 2 条数据）      |

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    print(cursor.rowcount)
    print(cursor.description)
    columns = [desc[0] for desc in cursor.description]
    print(columns)
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：

```txt
2
(('id', 3, None, 11, 11, 0, False), ('name', 253, None, 200, 200, 0, False), ('age', 3, None, 11, 11, 0, True), ('email', 253, None, 400, 400, 0, True))
['id', 'name', 'age', 'email']
```

**5. cursor.scroll(value, mode='relative')：移动游标位置**

作用：在结果集中移动游标（用于重新获取已跳过的数据）。

参数：

- value：移动的行数（正数向前，负数向后）。
- mode：'relative'（相对当前位置，默认）或 'absolute'（相对结果集开头）。

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    print(list(cursor.fetchone()))
    cursor.scroll(-1)
    print(list(cursor.fetchone()))
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：

```txt
[1, 'Alice', 26, 'alice@example.com']
[1, 'Alice', 26, 'alice@example.com']
```

```python
import pymysql

# 创建连接
conn = pymysql.connect(
    host="localhost",
    user="root",
    port=3306,
    password="root",
    database="test",
    charset="utf8mb4",
    autocommit=False  # 关闭自动提交，手动控制事务
)

cursor = conn.cursor()

try:
    cursor.execute("select * from users")
    print(list(cursor.fetchone()))
    print(list(cursor.fetchone()))
except Exception as e:
    print(f"事务失败：{e}")
```

输出结果：
```txt
[1, 'Alice', 26, 'alice@example.com']
[2, 'Bob', 30, 'bob@example.com']
```

**6. cursor.close()：关闭游标**

作用：释放游标资源，使用上下文管理器（with）时自动关闭。

## 异常处理

pymysql 的错误通过 pymysql.MySQLError 及其子类抛出，常用异常包括：

| 异常类           | 说明                                    |
| ---------------- | --------------------------------------- |
| MySQLError       | 所有 MySQL 错误的基类                   |
| OperationalError | 连接 / 操作错误（如连接超时、权限不足） |
| ProgrammingError | SQL 语法错误、表不存在等编程错误        |
| IntegrityError   | 数据完整性错误（如违反唯一约束 UNIQUE） |

```python
from pymysql import MySQLError, IntegrityError

try:
    cursor.execute("INSERT INTO users (email) VALUES ('duplicate@example.com')")
    conn.commit()
except IntegrityError as e:
    print(f"数据重复：{e}")  # 捕获邮箱重复的错误
except MySQLError as e:
    print(f"MySQL 错误：{e}")  # 捕获其他 MySQL 错误
```

## 常用方法

| 场景                      | 推荐方法 / 组合                                  |
| ------------------------- | ------------------------------------------------ |
| 单条 SQL 执行（增删改查） | cursor.execute() + 参数化查询                    |
| 批量插入 / 更新           | cursor.executemany() + conn.commit()             |
| 小批量查询结果            | cursor.fetchall()（一次性获取）                  |
| 大批量查询结果            | cursor.fetchmany(size)（分页获取，避免内存溢出） |
| 事务控制                  | conn.commit() + conn.rollback()                  |
| 动态获取列名              | cursor.description                               |

## 连接

```python
import pymysql

# 1. 配置连接参数
config = {
    "host": "localhost",       # 数据库主机地址（默认localhost）
    "port": 3306,              # 端口（默认3306）
    "user": "root",            # 用户名
    "password": "root",  # 密码（替换为实际密码）
    "database": "test",     # 要连接的数据库名
    "charset": "utf8mb4"       # 字符集（支持中文和表情符号）
}

# 2. 建立连接（推荐使用上下文管理器 with，自动释放资源）
try:
    with pymysql.connect(**config) as conn:  # 建立连接
        with conn.cursor() as cursor:       # 获取游标（用于执行SQL）
            # 3. 执行SQL（示例：查询数据库版本）
            cursor.execute("SELECT VERSION()")
            # 4. 处理结果（fetchone() 获取单条结果）
            version = cursor.fetchone()
            print(f"MySQL 版本：{version[0]}")

except pymysql.MySQLError as e:
    print(f"连接/执行失败：{e}")
```

输出结果：

```txt
MySQL 版本：5.5.53
```

- **Connection 对象**：代表与 MySQL 服务器的连接，提供事务控制（commit/rollback）、游标创建等方法。
- **Cursor 对象**：用于执行 SQL 语句并获取结果，支持多种结果获取方式（fetchone/fetchall 等）。
- **上下文管理器（with 语句）**：自动调用 close() 方法释放连接和游标，避免手动关闭遗漏导致的资源泄露。

## CRUD 操作

CRUD（Create/Read/Update/Delete）是数据库的基本操作，以下通过示例演示如何用 Python 实现。\

### 创建（Create）

**在test库中创建表**

```python
import pymysql
from pymysql import MySQLError

def create_users_table():
    # 数据库连接配置（请根据实际环境修改）
    config = {
        "host": "localhost",       # 数据库主机地址
        "port": 3306,              # 端口号
        "user": "root",            # 数据库用户名
        "password": "root",  # 替换为你的数据库密码
        "database": "test",        # 目标数据库（test库）
        "charset": "utf8mb4"       # 字符集，支持中文和特殊符号
    }

try:
        # 建立数据库连接（使用上下文管理器自动释放资源）
        with pymysql.connect(**config) as conn:
            # 创建游标对象
            with conn.cursor() as cursor:
                # 定义创建users表的SQL语句
                create_table_sql = """
                CREATE TABLE IF NOT EXISTS users (
                    id INT PRIMARY KEY AUTO_INCREMENT,
                    name VARCHAR(50) NOT NULL,
                    age INT,
                    email VARCHAR(100) UNIQUE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                """
                # 执行SQL语句
                cursor.execute(create_table_sql)
                print("users表创建成功")

except MySQLError as e:
        print(f"创建表失败：{e}")
    except Exception as e:
        print(f"发生未知错误：{e}")

create_users_table()
```

输出结果：
```txt
users表创建成功
```

在数据库中查询：

```mysql
mysql> desc users;
+-------+--------------+------+-----+---------+----------------+
| Field | Type         | Null | Key | Default | Extra          |
+-------+--------------+------+-----+---------+----------------+
| id    | int(11)      | NO   | PRI | NULL    | auto_increment |
| name  | varchar(50)  | NO   |     | NULL    |                |
| age   | int(11)      | YES  |     | NULL    |                |
| email | varchar(100) | YES  | UNI | NULL    |                |
+-------+--------------+------+-----+---------+----------------+
4 rows in set (0.00 sec)
```

**创建users表中的数据**

```python
import pymysql
from pymysql import MySQLError

def create_users_data():
    # 数据库连接配置（请根据实际环境修改）
    config = {
        "host": "localhost",       # 数据库主机地址
        "port": 3306,              # 端口号
        "user": "root",            # 数据库用户名
        "password": "root",  # 替换为你的数据库密码
        "database": "test",        # 目标数据库（test库）
        "charset": "utf8mb4"       # 字符集，支持中文和特殊符号
    }

try:
        with pymysql.connect(**config) as conn:
            with conn.cursor() as cursor:
                # 1. 单条插入
                sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
                # 使用参数化查询（%s 为占位符），避免 SQL 注入
                cursor.execute(sql, ("Alice", 25, "alice@example.com"))

# 2. 批量插入（效率高于单条多次插入）
                users = [
                    ("Bob", 30, "bob@example.com"),
                    ("Charlie", 35, "charlie@example.com")
                ]
                cursor.executemany(sql, users)  # executemany 支持批量操作

# 提交事务（插入/更新/删除需手动提交，查询无需）
                conn.commit()
                print(f"插入成功，影响行数：{cursor.rowcount}")  # rowcount 为受影响行数

except pymysql.MySQLError as e:
        conn.rollback()  # 出错时回滚事务
        print(f"插入失败：{e}")

create_users_data()
```

输出结果：
```txt
插入成功，影响行数：2
```

在数据库中查询：

```mysql
mysql> select * from users;
+----+---------+------+---------------------+
| id | name    | age  | email               |
+----+---------+------+---------------------+
|  1 | Alice   |   25 | alice@example.com   |
|  2 | Bob     |   30 | bob@example.com     |
|  3 | Charlie |   35 | charlie@example.com |
+----+---------+------+---------------------+
3 rows in set (0.00 sec)
```

**注意事项：**

- 必须使用 **参数化查询**（%s 占位符），而非字符串拼接（如 f"VALUES ('{name}')"），否则可能遭受 SQL 注入攻击。
- 插入 / 更新 / 删除操作需调用 conn.commit() 提交事务，否则修改不会生效；查询操作无需提交。

### 查询（Read）

```python
import pymysql
from pymysql import MySQLError

def select_users():
    # 数据库连接配置（请根据实际环境修改）
    config = {
        "host": "localhost",       # 数据库主机地址
        "port": 3306,              # 端口号
        "user": "root",            # 数据库用户名
        "password": "root",  # 替换为你的数据库密码
        "database": "test",        # 目标数据库（test库）
        "charset": "utf8mb4"       # 字符集，支持中文和特殊符号
    }

try:
        with pymysql.connect(**config) as conn:
            with conn.cursor(pymysql.cursors.DictCursor) as cursor:  # 返回字典格式（键为列名）
                # 1. 简单查询
                sql = "SELECT id, name, age FROM users WHERE age > %s"
                cursor.execute(sql, (28,))  # 查询年龄大于28的用户

# 2. 获取结果
                # fetchone()：获取第一条结果
                first_user = cursor.fetchone()
                print("第一条用户：", first_user)  # 输出：{'id': 2, 'name': 'Bob', 'age': 30}

# fetchall()：获取剩余所有结果
                remaining_users = cursor.fetchall()
                print("剩余用户：", remaining_users)

# 3. 分页查询（避免一次性加载大量数据）
                page_size = 10
                page_num = 1
                offset = (page_num - 1) * page_size
                sql = "SELECT * FROM users LIMIT %s OFFSET %s"
                cursor.execute(sql, (page_size, offset))
                page_users = cursor.fetchall()
                print(f"第{page_num}页用户：", page_users)

except pymysql.MySQLError as e:
        print(f"查询失败：{e}")

select_users()
```

输出结果：
```txt
第一条用户： {'id': 2, 'name': 'Bob', 'age': 30}
剩余用户： [{'id': 3, 'name': 'Charlie', 'age': 35}]
第1页用户： [
{'id': 1, 'name': 'Alice', 'age': 25, 'email': 'alice@example.com'}, 
{'id': 2, 'name': 'Bob', 'age': 30, 'email': 'bob@example.com'}, 
{'id': 3, 'name': 'Charlie', 'age': 35, 'email': 'charlie@example.com'}
]
```

**注意事项：**

- cursor.fetchone()：获取下一条结果（返回单条数据，无数据时返回 None）。
- cursor.fetchall()：获取所有剩余结果（返回列表，适合小批量数据）。
- cursor.fetchmany(size)：获取指定数量的结果（适合分页）。
- 游标默认返回元组（如 (2, 'Bob', 30)），通过 cursorclass=pymysql.cursors.DictCursor 可返回字典（键为列名），更直观。

### 更新（Update）

```python
import pymysql
from pymysql import MySQLError

def update_users():
    # 数据库连接配置（请根据实际环境修改）
    config = {
        "host": "localhost",       # 数据库主机地址
        "port": 3306,              # 端口号
        "user": "root",            # 数据库用户名
        "password": "root",  # 替换为你的数据库密码
        "database": "test",        # 目标数据库（test库）
        "charset": "utf8mb4"       # 字符集，支持中文和特殊符号
    }

try:
        with pymysql.connect(**config) as conn:
            with conn.cursor() as cursor:
                # 更新 Alice 的年龄为 26
                sql = "UPDATE users SET age = %s WHERE name = %s"
                cursor.execute(sql, (26, "Alice"))
                conn.commit()
                print(f"更新成功，影响行数：{cursor.rowcount}")  # 若 Alice 存在，输出 1

except pymysql.MySQLError as e:
        conn.rollback()
        print(f"更新失败：{e}")

update_users()
```

输出结果：

```txt
更新成功，影响行数：1
```

在数据库查询：

```mysql
mysql> select * from users;
+----+---------+------+---------------------+
| id | name    | age  | email               |
+----+---------+------+---------------------+
|  1 | Alice   |   26 | alice@example.com   |
|  2 | Bob     |   30 | bob@example.com     |
|  3 | Charlie |   35 | charlie@example.com |
+----+---------+------+---------------------+
3 rows in set (0.00 sec)
```

### 删除（Delete）

```python
import pymysql
from pymysql import MySQLError

def delete_users():
    # 数据库连接配置（请根据实际环境修改）
    config = {
        "host": "localhost",       # 数据库主机地址
        "port": 3306,              # 端口号
        "user": "root",            # 数据库用户名
        "password": "root",  # 替换为你的数据库密码
        "database": "test",        # 目标数据库（test库）
        "charset": "utf8mb4"       # 字符集，支持中文和特殊符号
    }

try:
        with pymysql.connect(**config) as conn:
            with conn.cursor() as cursor:
                # 删除邮箱为 charlie@example.com 的用户
                sql = "DELETE FROM users WHERE email = %s"
                cursor.execute(sql, ("charlie@example.com",))
                conn.commit()
                print(f"删除成功，影响行数：{cursor.rowcount}")

except pymysql.MySQLError as e:
        conn.rollback()
        print(f"删除失败：{e}")

delete_users()
```

输出结果：

```txt
删除成功，影响行数：1
```

## 事务处理

事务（Transaction）是一组原子性的 SQL 操作，要么全部成功，要么全部失败（ACID 特性）。

适用于多步操作需保证一致性的场景（如转账：扣减余额 + 增加对方余额，两步必须同时成功）。

```python
try:
    with pymysql.connect(** config) as conn:
        # 关闭自动提交（默认自动提交，事务需手动控制）
        conn.autocommit(False)
        
        with conn.cursor() as cursor:
            # 步骤1：扣减用户A的余额
            cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
            # 步骤2：增加用户B的余额
            cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
            
            # 若两步均成功，提交事务
            conn.commit()
            print("转账成功")

except pymysql.MySQLError as e:
    # 若任意步骤失败，回滚事务（恢复到操作前状态）
    conn.rollback()
    print(f"转账失败，已回滚：{e}")
```

**注意事项：**

- conn.autocommit(False)：关闭自动提交（默认 True，每条 SQL 自动提交）。
- conn.commit()：提交事务（所有操作生效）。
- conn.rollback()：回滚事务（所有操作失效，恢复到事务开始前状态）。

## 连接池

在高并发场景（如 Web 服务）中，频繁创建和关闭数据库连接会导致性能损耗。**连接池** 可预先创建一定数量的连接，供请求复用，减少连接开销。

常用连接池库：DBUtils（需配合 PyMySQL 使用）。

安装：

```cmd
pip install dbutils
```

示例：
```python
from dbutils.pooled_db import PooledDB
import pymysql

# 配置连接池
POOL = PooledDB(
    creator=pymysql,        # 使用的数据库驱动
    maxconnections=10,      # 连接池最大连接数
    mincached=2,            # 初始化时连接池中的空闲连接数
    maxcached=5,            # 连接池最大空闲连接数
    maxshared=3,            # 共享连接的最大数量（0表示所有连接都是专用的）
    blocking=True,          # 当连接池无可用连接时，是否阻塞等待（True：等待；False：报错）
    maxusage=None,          # 单个连接的最大使用次数（None表示无限制）
    setsession=[],          # 连接建立后执行的SQL命令（如设置时区）
    ping=1,                 # 检查连接有效性的方式（1：每次请求前 ping 服务器）
    # 数据库连接参数（同普通连接）
    host="localhost",
    user="root",
    password="root",
    database="test",
    charset="utf8mb4"
)

# 使用连接池获取连接
def query_users():
    try:
        # 从连接池获取连接（无需手动创建）
        conn = POOL.connection()
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("SELECT name, age FROM users")
            return cursor.fetchall()
    except pymysql.MySQLError as e:
        print(f"查询失败：{e}")
    finally:
        # 归还连接到池（非关闭连接）
        conn.close()

# 测试连接池
users = query_users()
print(users)
```

输出结果：

```txt
[{'name': 'Alice', 'age': 26}, {'name': 'Bob', 'age': 30}]
```

在数据库中查询：

```mysql
mysql> select * from users;
+----+-------+------+-------------------+
| id | name  | age  | email             |
+----+-------+------+-------------------+
|  1 | Alice |   26 | alice@example.com |
|  2 | Bob   |   30 | bob@example.com   |
+----+-------+------+-------------------+
2 rows in set (0.00 sec)
```

优势：

- 减少连接创建 / 关闭的开销，提升高并发场景性能。
- 控制最大连接数，避免数据库因连接过多而崩溃。

## 常见问题

| 问题       | 原因与解决方案                                               |
| ---------- | ------------------------------------------------------------ |
| 连接超时   | MySQL 服务器默认会关闭长时间闲置的连接，可在连接参数中添加 connect_timeout=30（超时时间 30 秒），或在连接池配置 ping=1 保持连接活性。 |
| 中文乱码   | 连接时指定 charset="utf8mb4"，并确保数据库表的字符集为 utf8mb4（CREATE TABLE ... CHARSET=utf8mb4）。 |
| 端口被占用 | 检查 MySQL 服务是否启动（netstat -tlnp grep 3306），或者检查是否修改过默认端口。 |
| 权限不足   | 登录 MySQL 服务器，为用户授予权限：GRANT ALL PRIVILEGES ON test_db.* TO 'user'@'localhost' IDENTIFIED BY 'password'; |

**参数化查询必须用 %s**：无论参数类型（字符串、数字），均用 %s 作为占位符，禁止用字符串拼接（如 f"VALUES ('{name}')"），否则会导致 SQL 注入。

**游标类型选择**：简单场景用默认游标（元组），需要字段名时用 DictCursor（字典）。

**连接与游标释放**：优先使用上下文管理器（with 语句），自动关闭资源，避免连接泄漏。

**事务提交**：插入 / 更新 / 删除操作需手动调用 conn.commit()，查询操作无需提交。