在实际数据库开发中，事务管理是保障数据一致性的核心机制（如转账时 “扣款” 与 “到账” 需同时成功或同时失败），而数据备份恢复是应对数据丢失的重要保障。本章基于python_db数据库，系统讲解 Python 中的事务控制方法、事务 ACID 特性，以及数据备份（导出）与恢复（导入）的实操流程，通过规整格式和案例演示，帮助你掌握数据库稳定性保障的关键技能。

6.1 数据库事务的基础概念

6.1.1 事务的定义

事务（Transaction）是数据库中一组不可分割的操作单元，这组操作要么全部执行成功，要么全部执行失败，不存在 “部分成功” 的情况。例如：

• 电商下单：需同时执行 “扣减库存” 和 “生成订单” 两个操作，若其中一个失败，整个流程需回退。

• 银行转账：需同时执行 “转出账户扣款” 和 “转入账户到账”，若任一操作失败，需恢复初始状态。

6.1.2 事务的 ACID 特性（核心原则）

事务必须满足 ACID 四个特性，才能确保数据一致性，具体如下表所示：

6.1.3 MySQL 的事务默认行为

• MySQL 中，InnoDB存储引擎支持事务（MyISAM不支持），而python_db数据库的users表默认使用InnoDB。

• 默认情况下，MySQL 开启 “自动提交” 模式（autocommit = ON），即每条INSERT/UPDATE/DELETE语句会自动作为一个独立事务提交。

• 实际开发中，需关闭自动提交，手动控制事务（如多步操作组成一个事务）。

6.2 Python 中的事务控制流程

Python 通过数据库连接对象（connection）实现事务控制，核心步骤为 “关闭自动提交→执行事务操作→提交 / 回滚事务”，具体流程如下：

6.2.1 事务控制核心方法

6.2.2 单事务多操作案例（转账模拟）

以 “用户 A 向用户 B 转账 100 元” 为例，需执行两步操作：

1. 用户 A（name='张三更新'）余额减 100

2. 用户 B（name='李四'）余额加 100

若任一操作失败，需回滚整个事务，确保数据一致性。

前提：为users表新增balance字段（余额）

# 执行一次，为users表新增余额字段（默认值1000）

def add_balance_field():

connection = None

try:

connection = mysql.connector.connect(

host='localhost',

user='root',

password='your_password',

database='python_db'

)

if connection.is_connected():

cursor = connection.cursor()

cursor.execute("ALTER TABLE users ADD COLUMN balance INT DEFAULT 1000")

connection.commit()

print("已为users表新增balance字段（默认余额1000）")

except Error as e:

print(f"新增字段出错：{e}")

finally:

if connection and connection.is_connected():

cursor.close()

connection.close()

add_balance_field()

Python 事务控制实现（转账）

import mysql.connector

from mysql.connector import Error

def transfer_money(from_name, to_name, amount):

"""

转账事务函数

:param from_name: 转出用户姓名

:param to_name: 转入用户姓名

:param amount: 转账金额（正数）

"""

connection = None

try:

# 1. 数据库连接并关闭自动提交

connection = mysql.connector.connect(

host='localhost',

user='root',

password='your_password',

database='python_db'

)

connection.autocommit = False # 关闭自动提交，开启手动事务

cursor = connection.cursor()

# 2. 转账前查询：确认转出用户余额足够

print("=== 转账前账户状态 ===")

# 查询转出用户

cursor.execute("SELECT name, balance FROM users WHERE name = %s", (from_name,))

from_user = cursor.fetchone()

# 查询转入用户

cursor.execute("SELECT name, balance FROM users WHERE name = %s", (to_name,))

to_user = cursor.fetchone()

# 验证用户是否存在、余额是否足够

if not from_user:

raise Exception(f"转出用户'{from_name}'不存在")

if not to_user:

raise Exception(f"转入用户'{to_name}'不存在")

if from_user[1] < amount:

raise Exception(f"转出用户'{from_name}'余额不足（当前余额：{from_user[1]}）")

# 打印转账前状态

field_names = ["用户姓名", "当前余额"]

print("|".join([f"{name:<10}" for name in field_names]))

print("-" * (10 * len(field_names) + len(field_names) - 1))

print("|".join([f"{from_user[0]:<10}", f"{from_user[1]:<10}"]))

print("|".join([f"{to_user[0]:<10}", f"{to_user[1]:<10}"]))

# 3. 执行转账操作（两步操作组成一个事务）

# 步骤1：转出用户余额减amount

cursor.execute(

"UPDATE users SET balance = balance - %s WHERE name = %s",

(amount, from_name)

)

# 步骤2：转入用户余额加amount

cursor.execute(

"UPDATE users SET balance = balance + %s WHERE name = %s",

(amount, to_name)

)

# 4. 提交事务（所有操作成功，永久保存）

connection.commit()

print(f"\n转账成功！{from_name}向{to_name}转账{amount}元")

# 5. 转账后验证

print("\n=== 转账后账户状态 ===")

cursor.execute("SELECT name, balance FROM users WHERE name IN (%s, %s)", (from_name, to_name))

post_transfer = cursor.fetchall()

print("|".join([f"{name:<10}" for name in field_names]))

print("-" * (10 * len(field_names) + len(field_names) - 1))

for user in post_transfer:

print("|".join([f"{user[0]:<10}", f"{user[1]:<10}"]))

except Exception as e:

# 6. 事务回滚（任一操作失败，撤销所有修改）

print(f"\n转账失败：{e}")

if connection and connection.is_connected():

connection.rollback()

print("已执行事务回滚，恢复到转账前状态")

finally:

# 7. 关闭连接

if connection and connection.is_connected():

cursor.close()

connection.close()

print("\nMySQL连接已关闭")

# 执行转账（张三更新向李四转账100元）

if __name__ == "__main__":

transfer_money(from_name="张三更新", to_name="李四", amount=100)

运行结果（成功场景，规整对齐）

=== 转账前账户状态 ===

用户姓名 |当前余额

-------------------------

张三更新 |1000

李四 |1000

转账成功！张三更新向李四转账100元

=== 转账后账户状态 ===

用户姓名 |当前余额

-------------------------

张三更新 |900

李四 |1100

MySQL连接已关闭

运行结果（失败场景：余额不足）

=== 转账前账户状态 ===

用户姓名 |当前余额

-------------------------

张三更新 |900

李四 |1100

转账失败：转出用户'张三更新'余额不足（当前余额：900）

已执行事务回滚，恢复到转账前状态

MySQL连接已关闭

6.3 事务隔离级别（避免并发问题）

当多个事务同时执行时，可能出现脏读、不可重复读、幻读等并发问题，MySQL 通过 “隔离级别” 控制事务间的可见性，解决这些问题。

6.3.1 常见并发问题

6.3.2 MySQL 的四种隔离级别

MySQL 支持四种隔离级别（从低到高，并发性能递减，数据一致性递增）：

6.3.3 Python 设置隔离级别示例

def set_isolation_level(level):

"""设置事务隔离级别并验证"""

connection = None

try:

connection = mysql.connector.connect(

host='localhost',

user='root',

password='your_password',

database='python_db'

)

# 设置隔离级别

connection.isolation_level = level

# 验证设置结果

current_level = connection.isolation_level

print(f"事务隔离级别设置成功，当前级别：{current_level}")

except Error as e:

print(f"设置隔离级别出错：{e}")

finally:

if connection and connection.is_connected():

connection.close()

# 示例：设置为“读已提交”级别

if __name__ == "__main__":

set_isolation_level('READ COMMITTED')

运行结果

事务隔离级别设置成功，当前级别：READ COMMITTED

6.4 数据备份（导出）与恢复（导入）

数据备份是防止数据丢失的关键手段，Python 可通过SELECT ... INTO OUTFILE实现数据导出（备份），通过LOAD DATA INFILE实现数据导入（恢复）。

6.4.1 数据备份（导出为 CSV 文件）

将users表数据导出为 CSV 文件（逗号分隔，便于 Excel 打开和长期存储），步骤如下：

Python 实现数据导出

def backup_table_to_csv(table_name, output_path):

"""

数据表导出为CSV文件

:param table_name: 表名（如'users'）

:param output_path: 输出文件路径（如'./users_backup.csv'）

"""

connection = None

try:

connection = mysql.connector.connect(

host='localhost',

user='root',

password='your_password',

database='python_db'

)

if connection.is_connected():

cursor = connection.cursor()

# 导出SQL：字段用逗号分隔，行用换行分隔，忽略表头

backup_query = f"""

SELECT * FROM {table_name}

INTO OUTFILE '{output_path}'

FIELDS TERMINATED BY ',' # 字段分隔符

ENCLOSED BY '"' # 字段值用双引号包裹（避免含逗号的字符串拆分）

LINES TERMINATED BY '\n' # 行分隔符

"""

cursor.execute(backup_query)

print(f"数据表'{table_name}'备份成功，文件路径：{output_path}")

# 验证备份文件行数（与表记录数对比）

cursor.execute(f"SELECT COUNT(*) FROM {table_name}")

table_count = cursor.fetchone()[0]

with open(output_path, 'r', encoding='utf-8') as f:

backup_count = len(f.readlines())

print(f"验证：表记录数={table_count}，备份文件行数={backup_count}（一致则备份完整）")

except Error as e:

print(f"数据备份出错：{e}")

# 常见错误：MySQL安全策略禁止导出到指定路径，需修改my.cnf配置（添加secure-file-priv=''）

finally:

if connection and connection.is_connected():

cursor.close()

connection.close()

# 执行备份（导出users表到当前目录的users_backup.csv）

if __name__ == "__main__":

backup_table_to_csv(table_name='users', output_path='./users_backup.csv')

运行结果（成功场景）

数据表'users'备份成功，文件路径：./users_backup.csv

验证：表记录数=5，备份文件行数=5（一致则备份完整）

6.4.2 数据恢复（从 CSV 文件导入）

假设users表数据意外丢失，可通过备份的 CSV 文件恢复数据，步骤如下：

前提：清空users表（模拟数据丢失）

def clear_table(table_name):

"""清空表数据（仅用于测试恢复场景）"""

connection = None

try:

connection = mysql.connector.connect(

host='localhost',

user='root',

password='your_password',

database='python_db'

)

connection.autocommit = False

cursor = connection.cursor()

confirm = input(f"确认清空表'{table_name}'的所有数据？（y/n）：")

if confirm.lower() == 'y':

cursor.execute(f"DELETE FROM {table_name}")

connection.commit()

print(f"表'{table_name}'数据已清空")

else:

print("已取消清空操作")

except Error as e:

print(f"清空表出错：{e}")

if connection:

connection.rollback()

finally:

if connection and connection.is_connected():

cursor.close()

connection.close()

# 执行清空（测试用）

# clear_table</doubaocanvas>