本文目录导读：
我来详细说明Python中数据库事务的控制方法,包含多个实际案例。
基础事务控制（以SQLite为例）
import sqlite3
from contextlib import contextmanager
class DatabaseManager:
def __init__(self, db_path):
self.db_path = db_path
self.conn = None
def __enter__(self):
self.conn = sqlite3.connect(self.db_path)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.conn:
self.conn.close()
@contextmanager
def transaction(self):
"""事务上下文管理器"""
try:
yield self.conn
self.conn.commit()
print("事务提交成功")
except Exception as e:
self.conn.rollback()
print(f"事务回滚: {e}")
raise
# 使用示例
def bank_transfer_example():
"""银行转账示例（含事务控制）"""
with DatabaseManager('bank.db') as db:
# 创建表和测试数据
db.conn.execute('''CREATE TABLE IF NOT EXISTS accounts
(id INTEGER PRIMARY KEY, name TEXT, balance REAL)''')
db.conn.execute("INSERT OR IGNORE INTO accounts VALUES (1, '张三', 1000)")
db.conn.execute("INSERT OR IGNORE INTO accounts VALUES (2, '李四', 500)")
db.conn.commit()
# 使用事务进行转账
with db.transaction():
cursor = db.conn.cursor()
# 检查余额
cursor.execute("SELECT balance FROM accounts WHERE id=1")
balance_zhang = cursor.fetchone()[0]
if balance_zhang < 200:
raise ValueError("余额不足")
# 执行转账
cursor.execute("UPDATE accounts SET balance = balance - 200 WHERE id=1")
cursor.execute("UPDATE accounts SET balance = balance + 200 WHERE id=2")
print("转账成功：张三 -> 李四 200元")
# 运行示例
try:
bank_transfer_example()
except Exception as e:
print(f"转账失败: {e}")
MySQL事务控制（使用mysql-connector）
import mysql.connector
from mysql.connector import Error
class MySQLTransactionManager:
def __init__(self, host, user, password, database):
self.connection_config = {
'host': host,
'user': user,
'password': password,
'database': database
}
self.connection = None
def connect(self):
"""建立数据库连接"""
try:
self.connection = mysql.connector.connect(**self.connection_config)
# 关闭自动提交，手动控制事务
self.connection.autocommit = False
print("数据库连接成功")
except Error as e:
print(f"连接失败: {e}")
raise
def disconnect(self):
"""关闭连接"""
if self.connection and self.connection.is_connected():
self.connection.close()
print("数据库连接已关闭")
def execute_transaction(self, operations):
"""
执行事务操作
:param operations: 操作列表，每个元素为(sql, params)元组
"""
cursor = None
try:
cursor = self.connection.cursor()
# 执行所有操作
for sql, params in operations:
cursor.execute(sql, params)
# 提交事务
self.connection.commit()
print("事务提交成功")
return True
except Error as e:
print(f"事务执行失败: {e}")
# 回滚事务
if self.connection:
self.connection.rollback()
print("事务已回滚")
return False
finally:
if cursor:
cursor.close()
# 使用示例
def order_processing_example():
"""订单处理事务示例"""
manager = MySQLTransactionManager(
host='localhost',
user='root',
password='password',
database='shop'
)
try:
manager.connect()
# 定义事务操作（订单创建、库存更新、账户扣款）
operations = [
("INSERT INTO orders (user_id, product_id, quantity, total_price) VALUES (%s, %s, %s, %s)",
(1, 100, 2, 199.98)),
("UPDATE products SET stock = stock - %s WHERE id = %s AND stock >= %s",
(2, 100, 2)), # 减库存
("UPDATE user_balance SET balance = balance - %s WHERE user_id = %s AND balance >= %s",
(199.98, 1, 199.98)) # 扣款
]
success = manager.execute_transaction(operations)
if success:
print("订单创建成功")
else:
print("订单创建失败")
finally:
manager.disconnect()
# 调用示例
# order_processing_example()
使用ORM（SQLAlchemy）的事务控制
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager
Base = declarative_base()
# 定义模型
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(100))
stock = Column(Integer, default=0)
price = Column(Float)
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
product_id = Column(Integer)
quantity = Column(Integer)
total_price = Column(Float)
class SQLAlchemyTransactionDemo:
def __init__(self, database_url='sqlite:///shop.db'):
self.engine = create_engine(database_url)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
@contextmanager
def session_scope(self):
"""提供事务范围的会话上下文"""
session = self.Session()
try:
yield session
session.commit()
print("事务提交成功")
except Exception as e:
session.rollback()
print(f"事务回滚: {e}")
raise
finally:
session.close()
def create_order_with_transaction(self, product_id, quantity, user_id=1):
"""在事务中创建订单"""
with self.session_scope() as session:
# 检查产品库存
product = session.query(Product).filter_by(id=product_id).first()
if not product:
raise ValueError("产品不存在")
if product.stock < quantity:
raise ValueError(f"库存不足，当前库存: {product.stock}")
# 计算总价
total_price = product.price * quantity
# 创建订单
order = Order(
product_id=product_id,
quantity=quantity,
total_price=total_price
)
session.add(order)
# 更新库存
product.stock -= quantity
# 这里可以添加更多事务操作，如账户扣款等
print(f"订单创建成功: 产品{product.name}, 数量{quantity}, 总价{total_price}")
# 使用示例
def orm_transaction_example():
demo = SQLAlchemyTransactionDemo()
# 初始化测试数据
with demo.session_scope() as session:
# 检查是否已存在数据
if not session.query(Product).first():
product = Product(name="笔记本电脑", stock=10, price=5000.0)
session.add(product)
# 创建订单（自动在事务中执行）
try:
demo.create_order_with_transaction(product_id=1, quantity=2)
# 尝试库存不足的情况
demo.create_order_with_transaction(product_id=1, quantity=20)
except ValueError as e:
print(f"业务错误: {e}")
# 运行示例
orm_transaction_example()
事务隔离级别控制
import sqlite3
from threading import Thread, Lock
import time
class IsolationLevelDemo:
def __init__(self, db_path='isolation.db'):
self.db_path = db_path
self.lock = Lock()
self.init_database()
def init_database(self):
"""初始化数据库和测试数据"""
conn = sqlite3.connect(self.db_path)
conn.execute('''CREATE TABLE IF NOT EXISTS inventory
(id INTEGER PRIMARY KEY, item TEXT, quantity INTEGER)''')
# 清空并插入测试数据
conn.execute("DELETE FROM inventory")
conn.execute("INSERT INTO inventory VALUES (1, '商品A', 100)")
conn.commit()
conn.close()
def transaction_with_isolation(self, isolation_level='IMMEDIATE'):
"""
使用不同隔离级别的事务
:param isolation_level: DEFERRED, IMMEDIATE, EXCLUSIVE
"""
conn = sqlite3.connect(self.db_path)
# 设置事务类型
conn.execute(f"BEGIN {isolation_level}")
try:
cursor = conn.cursor()
# 读取数据
cursor.execute("SELECT quantity FROM inventory WHERE id=1")
current_qty = cursor.fetchone()[0]
print(f"当前库存: {current_qty}")
# 模拟业务处理
time.sleep(2)
# 更新数据
new_qty = current_qty - 10
cursor.execute("UPDATE inventory SET quantity=? WHERE id=1", (new_qty,))
conn.commit()
print(f"事务提交，新库存: {new_qty}")
return True
except Exception as e:
conn.rollback()
print(f"事务回滚: {e}")
return False
finally:
conn.close()
def concurrent_transaction_test(self):
"""并发事务测试"""
def worker(worker_id):
print(f"工作线程 {worker_id} 开始")
result = self.transaction_with_isolation('IMMEDIATE')
print(f"工作线程 {worker_id} {'成功' if result else '失败'}")
# 创建两个并发线程
threads = [
Thread(target=worker, args=(1,)),
Thread(target=worker, args=(2,))
]
# 启动线程
for t in threads:
t.start()
# 等待线程完成
for t in threads:
t.join()
# 运行并发测试
def run_isolation_test():
demo = IsolationLevelDemo()
print("="*50)
print("并发事务测试")
print("="*50)
# 使用EXCLUSIVE模式避免死锁
demo.concurrent_transaction_test()
# 检查最终结果
conn = sqlite3.connect('isolation.db')
cursor = conn.execute("SELECT * FROM inventory")
print(f"\n最终库存状态: {cursor.fetchall()}")
conn.close()
# 启动测试
# run_isolation_test()
最佳实践示例
import sqlite3
import logging
from typing import List, Tuple, Any
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RobustTransactionManager:
"""
健壮的事务管理器，包含重试机制和监控
"""
def __init__(self, db_path: str, max_retries: int = 3):
self.db_path = db_path
self.max_retries = max_retries
self.transaction_stats = {
'success': 0,
'failed': 0,
'retried': 0
}
def execute_transaction(self, operations: List[Tuple[str, Tuple]],
isolation_level: str = 'IMMEDIATE'):
"""
执行事务，包含重试机制
"""
for attempt in range(self.max_retries):
conn = None
try:
conn = sqlite3.connect(self.db_path)
# 开始事务
conn.execute(f"BEGIN {isolation_level}")
cursor = conn.cursor()
# 执行所有操作
for sql, params in operations:
cursor.execute(sql, params)
# 提交事务
conn.commit()
self.transaction_stats['success'] += 1
logger.info(f"事务执行成功 (尝试 {attempt + 1})")
return True
except sqlite3.OperationalError as e:
# 可重试的错误
if 'database is locked' in str(e):
self.transaction_stats['retried'] += 1
wait_time = (attempt + 1) * 0.5 # 递增等待时间
logger.warning(f"数据库锁定，等待 {wait_time}秒后重试")
time.sleep(wait_time)
continue
else:
if conn:
conn.rollback()
self.transaction_stats['failed'] += 1
logger.error(f"事务执行失败: {e}")
return False
except Exception as e:
if conn:
conn.rollback()
self.transaction_stats['failed'] += 1
logger.error(f"未预期的错误: {e}")
return False
finally:
if conn:
conn.close()
return False
def print_stats(self):
"""打印事务统计信息"""
print(f"\n事务统计:")
print(f"成功: {self.transaction_stats['success']}")
print(f"失败: {self.transaction_stats['failed']}")
print(f"重试: {self.transaction_stats['retried']}")
# 完整示例：订单系统
class OrderSystem:
def __init__(self, db_path='orders.db'):
self.tm = RobustTransactionManager(db_path)
self.init_tables()
def init_tables(self):
"""初始化数据库表"""
conn = sqlite3.connect(self.tm.db_path)
conn.executescript('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER,
quantity INTEGER,
total_price REAL,
order_date TIMESTAMP,
status TEXT DEFAULT 'pending'
);
CREATE TABLE IF NOT EXISTS customer_balance (
customer_id INTEGER PRIMARY KEY,
balance REAL DEFAULT 0
);
''')
conn.commit()
conn.close()
def place_order(self, customer_id: int, product_id: int, quantity: int):
"""下订单（完整事务控制）"""
# 准备事务操作
operations = [
# 1. 检查并扣减库存
("""UPDATE products
SET stock = stock - ?
WHERE id = ? AND stock >= ?""",
(quantity, product_id, quantity)),
# 2. 查询产品价格
("SELECT price FROM products WHERE id = ?", (product_id,)),
# 注意：这里需要返回价格，但事务中不能直接获取返回值
# 在实际应用中，应该先查询再更新
]
conn = sqlite3.connect(self.tm.db_path)
try:
conn.execute("BEGIN IMMEDIATE")
cursor = conn.cursor()
# 先查询产品信息
cursor.execute("SELECT price, stock FROM products WHERE id=?", (product_id,))
product = cursor.fetchone()
if not product:
raise ValueError("产品不存在")
price, stock = product
if stock < quantity:
raise ValueError(f"库存不足 (可用: {stock}, 需要: {quantity})")
# 扣减库存
cursor.execute("UPDATE products SET stock = stock - ? WHERE id=?",
(quantity, product_id))
# 检查客户余额
cursor.execute("SELECT balance FROM customer_balance WHERE customer_id=?",
(customer_id,))
customer = cursor.fetchone()
if not customer:
raise ValueError("客户不存在")
total_price = price * quantity
balance = customer[0]
if balance < total_price:
raise ValueError(f"余额不足 (可用: {balance}, 需要: {total_price})")
# 扣减客户余额
cursor.execute("UPDATE customer_balance SET balance = balance - ? WHERE customer_id=?",
(total_price, customer_id))
# 创建订单
cursor.execute("""
INSERT INTO orders (product_id, quantity, total_price, order_date, status)
VALUES (?, ?, ?, ?, 'completed')
""", (product_id, quantity, total_price, datetime.now()))
conn.commit()
print(f"订单创建成功: 产品ID={product_id}, 数量={quantity}, 总价={total_price}")
return True
except Exception as e:
conn.rollback()
print(f"订单创建失败: {e}")
return False
finally:
conn.close()
# 使用示例
def main():
# 创建系统
system = OrderSystem()
# 初始化测试数据
conn = sqlite3.connect('orders.db')
conn.execute("INSERT OR IGNORE INTO products VALUES (1, '商品A', 100.0, 50)")
conn.execute("INSERT OR IGNORE INTO products VALUES (2, '商品B', 200.0, 30)")
conn.execute("INSERT OR IGNORE INTO customer_balance VALUES (1, 5000.0)")
conn.commit()
conn.close()
# 下单测试
print("="*50)
print("测试1: 正常下单")
system.place_order(1, 1, 2) # 客户1购买2个商品A
print("\n" + "="*50)
print("测试2: 库存不足")
system.place_order(1, 1, 100) # 尝试购买100个
print("\n" + "="*50)
print("测试3: 余额不足")
system.place_order(1, 2, 30) # 30 * 200 = 6000，余额只有5000
if __name__ == "__main__":
main()
- 原子性保证：所有操作要么全部成功，要么全部失败
- 使用上下文管理器：用
with语句确保事务正确关闭 - 错误处理：捕获异常并进行回滚
- 隔离级别选择：根据并发需求选择合适的隔离级别
- 重试机制：处理死锁和超时情况
- 监控统计：记录事务执行情况便于排查问题
这些示例展示了从简单到复杂的各种事务控制方式,你可以根据实际需求选择合适的方案。
