Python 如何处理数据库事务

数据库事务是指一组操作要么全部执行成功，要么全部回滚的过程。事务是确保数据库一致性的重要手段，特别是在处理需要多步操作的场景时，能够避免部分数据成功更新而部分数据失败的情况。本文将详细介绍什么是数据库事务，以及如何在 Python 中通过各种方式（例如，使用 sqlite3、MySQL、PostgreSQL 等数据库驱动）处理事务。

在这里插入图片描述

什么是数据库事务？

1. 事务的概念

数据库事务（transaction）是指一组操作，它们被看作是一个单一的逻辑单元，要么全部执行成功，要么全部执行失败。事务具有以下四个特性，简称为 ACID 特性：

原子性（Atomicity）：事务中的所有操作要么全部成功，要么全部失败并回滚。事务不能处于部分成功、部分失败的状态。
一致性（Consistency）：事务开始前和结束后，数据库的状态必须保持一致，任何违反数据库规则的操作都不能提交。
隔离性（Isolation）：事务之间彼此独立执行，事务的中间状态对于其他事务是不可见的。
持久性（Durability）：事务一旦提交，数据将被持久保存，即使发生系统故障也不会丢失。

2. 为什么需要事务？

事务对于保证数据库操作的可靠性和数据的一致性至关重要。以下是几个常见的使用场景：

多步骤的操作：比如，在电商平台中，用户购买商品时，涉及到扣除库存、创建订单和扣减用户账户余额等操作，这些操作必须作为一个整体来执行。如果其中某一个步骤失败，则需要回滚所有已经完成的步骤，确保系统数据不会不一致。
防止部分更新：在同一个操作中，假设有多个更新操作，一旦其中某个操作失败，应该取消其他操作的影响，保证数据库不会处于不完整的更新状态。
并发控制：多个事务同时进行时，事务的隔离性可以避免脏读、不可重复读和幻读等问题。

在 Python 中处理事务

在 Python 中处理数据库事务通常涉及使用数据库驱动（如 sqlite3、psycopg2、PyMySQL 等），这些驱动通常提供内置的事务管理功能。以下是一些常见数据库库和事务管理的示例。

1. 使用 SQLite 处理事务

SQLite 是一个轻量级的数据库，广泛用于小型应用和开发环境中。在 Python 中，sqlite3 模块提供了事务支持。

事务的基本操作

sqlite3 默认是在“自动提交”模式下运行的，这意味着每个 SQL 语句都是一个事务。但是，你可以手动控制事务的开始和提交。

import sqlite3# 连接到数据库
conn = sqlite3.connect('example.db')try:# 创建游标cursor = conn.cursor()# 开始事务（默认自动开始，显示调用 conn.commit() 或 conn.rollback() 来结束）cursor.execute("BEGIN")# 执行多个 SQL 语句cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Alice", 30))cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")# 提交事务conn.commit()
except Exception as e:# 出现错误时回滚事务conn.rollback()print(f"Transaction failed: {e}")
finally:# 关闭连接conn.close()

在这个例子中，我们执行了两条 SQL 语句——插入用户数据和更新账户余额。这两条语句被视为一个事务，只有当两者都成功时，才会提交到数据库。一旦其中任何一个操作失败，事务将回滚，取消已完成的操作。

自动提交模式

在 sqlite3 中，如果不手动控制事务，它将处于自动提交模式。这意味着每条 SQL 语句执行后都会立即提交。

如果你需要在同一个连接中执行多条语句并确保它们作为一个事务执行，则需要手动控制事务，使用 conn.commit() 和 conn.rollback()。

2. 使用 MySQL 处理事务

在 MySQL 中，我们可以使用 PyMySQL 或 MySQL Connector/Python 等库来处理事务。

使用 PyMySQL 处理事务

import pymysql# 连接到 MySQL 数据库
conn = pymysql.connect(host='localhost',user='root',password='password',db='test_db'
)try:# 创建游标cursor = conn.cursor()# 执行多个 SQL 语句cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ("Bob", 25))cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 2")# 提交事务conn.commit()
except Exception as e:# 如果有任何错误，回滚事务conn.rollback()print(f"Transaction failed: {e}")
finally:# 关闭连接conn.close()

在 PyMySQL 中，默认情况下事务是手动控制的。在执行完一组操作后，使用 conn.commit() 提交事务；如果发生错误，则使用 conn.rollback() 回滚事务。

使用 MySQL Connector/Python 处理事务

MySQL Connector 是 MySQL 官方提供的 Python 库。事务操作与 PyMySQL 类似。

import mysql.connector# 连接到数据库
conn = mysql.connector.connect(host='localhost',user='root',password='password',database='test_db'
)try:cursor = conn.cursor()# 执行多条 SQL 语句cursor.execute("INSERT INTO orders (order_id, user_id) VALUES (%s, %s)", (101, 1))cursor.execute("UPDATE inventory SET quantity = quantity - 1 WHERE product_id = 1")# 提交事务conn.commit()
except Exception as e:# 回滚事务conn.rollback()print(f"Transaction failed: {e}")
finally:conn.close()

在 MySQL 中，事务控制的基本步骤与 PyMySQL 类似，都是先执行 SQL 操作，然后手动提交或回滚事务。

3. 使用 PostgreSQL 处理事务

PostgreSQL 是一个功能强大的开源数据库，支持丰富的事务功能。在 Python 中，psycopg2 是最常用的与 PostgreSQL 交互的库。

使用 `psycopg2` 处理事务

import psycopg2# 连接到 PostgreSQL 数据库
conn = psycopg2.connect(dbname="test_db",user="user",password="password",host="localhost"
)try:cursor = conn.cursor()# 执行 SQL 操作cursor.execute("INSERT INTO employees (name, department) VALUES (%s, %s)", ("John", "HR"))cursor.execute("UPDATE payroll SET salary = salary + 500 WHERE employee_id = 3")# 提交事务conn.commit()
except Exception as e:# 事务失败时回滚conn.rollback()print(f"Transaction failed: {e}")
finally:conn.close()

psycopg2 中的事务处理也是手动控制的，事务的提交和回滚需要明确调用 conn.commit() 或 conn.rollback()。

4. Django 中的事务处理

如果你使用的是 Django 这种高级 Web 框架，事务管理会更加简洁。Django 自带事务管理功能，并且有两种主要的事务处理方式：自动管理和手动管理。

自动事务管理

在 Django 中，默认情况下，所有的数据库操作都在自动提交模式下运行。每次数据库操作都会立即提交。

手动事务管理

对于更复杂的事务，Django 提供了 transaction.atomic() 这个上下文管理器，允许你手动控制事务。

from django.db import transactiontry:with transaction.atomic():user = User.objects.create(username="alice")Account.objects.create(user=user, balance=1000)
except Exception as e:print(f"Transaction failed: {e}")

在这个示例中，transaction.atomic() 将确保 User 和 Account 的创建操作作为一个原子操作。如果任何一个操作失败，整个事务将回滚。

5. SQLAlchemy 中的事务处理

如果你使用 ORM 库如 SQLAlchemy，事务管理也是非常简洁的。SQLAlchemy 允许你在会话（session）级别管理事务。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Account  # 假设已经定义了 ORM 模型# 连接数据库
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()try:# 创建用户和账户new_user = User(name="Alice")session.add(new_user)new_account = Account(user=new_user, balance=1000)session.add(new_account)# 提交事务session.commit()
except Exception as e:# 事务回滚session.rollback()print(f"Transaction failed: {e}")
finally:session.close()