Python实现数据库事务回滚案例分享

数据库事务是确保数据一致性的重要机制，而事务回滚则是在出现错误时保证数据完整性的关键手段。本文将通过多个实际案例展示如何在Python中实现数据库事务回滚。

1. 基础概念回顾

什么是数据库事务？

事务是一组SQL操作的集合，这些操作要么全部成功执行，要么全部不执行，具有以下四个特性（ACID）：

原子性（Atomicity）：事务是不可分割的工作单位
一致性（Consistency）：事务执行前后数据库保持一致状态
隔离性（Isolation）：多个事务并发执行时互不干扰
持久性（Durability）：事务提交后对数据库的修改是永久的

事务回滚的应用场景

银行转账：扣款和入账必须同时成功或同时失败
订单处理：创建订单和扣减库存需要原子操作
批量数据处理：部分失败时回滚所有操作

2. 使用SQLite实现事务回滚

SQLite是Python内置的轻量级数据库，非常适合学习和演示。

案例1：基础事务回滚

import sqlite3
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SQLiteTransactionDemo:
    def __init__(self, db_path=':memory:'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建账户表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS accounts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                balance REAL NOT NULL DEFAULT 0.0
            )
        ''')
        
        # 插入测试数据
        cursor.execute('DELETE FROM accounts')
        cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Alice', 1000.0)")
        cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Bob', 500.0)")
        
        conn.commit()
        conn.close()
        logger.info("数据库初始化完成")
    
    def transfer_money_success(self, from_account, to_account, amount):
        """成功的转账操作"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            # 开始事务（SQLite默认开启事务）
            logger.info(f"开始转账: {from_account} -> {to_account}, 金额: {amount}")
            
            # 检查转出账户余额
            cursor.execute("SELECT balance FROM accounts WHERE name = ?", (from_account,))
            from_balance = cursor.fetchone()[0]
            
            if from_balance < amount:
                raise ValueError(f"余额不足: {from_account} 当前余额 {from_balance}")
            
            # 执行转账操作
            cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", 
                         (amount, from_account))
            cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", 
                         (amount, to_account))
            
            # 提交事务
            conn.commit()
            logger.info("转账成功，事务已提交")
            
            # 验证结果
            self.show_balances()
            
        except Exception as e:
            # 回滚事务
            conn.rollback()
            logger.error(f"转账失败，事务已回滚: {e}")
            raise
        finally:
            conn.close()
    
    def transfer_money_with_error(self, from_account, to_account, amount):
        """会失败的转账操作（演示回滚）"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            logger.info(f"尝试转账: {from_account} -> {to_account}, 金额: {amount}")
            
            # 执行转账操作
            cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", 
                         (amount, from_account))
            cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", 
                         (amount, to_account))
            
            # 模拟一个错误：比如向不存在的账户转账
            cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", 
                         (amount, 'Charlie'))  # Charlie不存在，影响行数为0
            
            # 检查影响的行数
            if cursor.rowcount == 0:
                raise ValueError("目标账户不存在")
            
            conn.commit()
            logger.info("转账成功")
            
        except Exception as e:
            conn.rollback()
            logger.error(f"转账失败，事务已回滚: {e}")
            # 验证回滚结果
            self.show_balances()
            raise
        finally:
            conn.close()
    
    def show_balances(self):
        """显示所有账户余额"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT name, balance FROM accounts")
        balances = cursor.fetchall()
        conn.close()
        
        logger.info("当前账户余额:")
        for name, balance in balances:
            logger.info(f"  {name}: {balance}")

# 演示基础事务回滚
def demo_basic_transaction():
    demo = SQLiteTransactionDemo()
    
    print("=== 初始状态 ===")
    demo.show_balances()
    
    print("\n=== 成功转账演示 ===")
    try:
        demo.transfer_money_success('Alice', 'Bob', 200.0)
    except Exception as e:
        print(f"操作失败: {e}")
    
    print("\n=== 失败转账演示（触发回滚）===")
    try:
        demo.transfer_money_with_error('Alice', 'Bob', 100.0)
    except Exception as e:
        print(f"操作失败: {e}")
    
    print("\n=== 最终状态（验证回滚）===")
    demo.show_balances()

if __name__ == "__main__":
    demo_basic_transaction()

案例2：批量操作的事务回滚

import sqlite3
import logging
from typing import List, Tuple

logger = logging.getLogger(__name__)

class BatchTransactionDemo:
    def __init__(self, db_path=':memory:'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """初始化订单和库存表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建商品库存表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS products (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                stock INTEGER NOT NULL DEFAULT 0,
                price REAL NOT NULL
            )
        ''')
        
        # 创建订单表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS orders (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_id INTEGER,
                quantity INTEGER,
                total_price REAL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (product_id) REFERENCES products (id)
            )
        ''')
        
        # 插入测试数据
        cursor.execute('DELETE FROM products')
        cursor.execute('DELETE FROM orders')
        cursor.execute("INSERT INTO products (name, stock, price) VALUES ('Laptop', 10, 999.99)")
        cursor.execute("INSERT INTO products (name, stock, price) VALUES ('Mouse', 50, 29.99)")
        cursor.execute("INSERT INTO products (name, stock, price) VALUES ('Keyboard', 30, 79.99)")
        
        conn.commit()
        conn.close()
        logger.info("订单系统数据库初始化完成")
    
    def create_order_batch(self, orders: List[Tuple[str, int]]) -> bool:
        """
        批量创建订单
        如果任何一个订单失败（如库存不足），所有订单都回滚
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            logger.info(f"开始批量创建 {len(orders)} 个订单")
            
            # 开始事务
            for product_name, quantity in orders:
                # 检查商品是否存在
                cursor.execute("SELECT id, stock, price FROM products WHERE name = ?", (product_name,))
                product = cursor.fetchone()
                
                if not product:
                    raise ValueError(f"商品不存在: {product_name}")
                
                product_id, stock, price = product
                
                # 检查库存是否充足
                if stock < quantity:
                    raise ValueError(f"库存不足: {product_name} (请求: {quantity}, 可用: {stock})")
                
                # 计算总价
                total_price = price * quantity
                
                # 扣减库存
                cursor.execute("UPDATE products SET stock = stock - ? WHERE id = ?", 
                             (quantity, product_id))
                
                # 创建订单
                cursor.execute("INSERT INTO orders (product_id, quantity, total_price) VALUES (?, ?, ?)",
                             (product_id, quantity, total_price))
                
                logger.info(f"订单创建成功: {product_name} x {quantity}")
            
            # 所有订单处理成功，提交事务
            conn.commit()
            logger.info("批量订单创建成功，事务已提交")
            return True
            
        except Exception as e:
            # 任何一个订单失败，回滚所有操作
            conn.rollback()
            logger.error(f"批量订单创建失败，事务已回滚: {e}")
            return False
        finally:
            conn.close()
    
    def show_inventory(self):
        """显示商品库存"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT name, stock, price FROM products")
        products = cursor.fetchall()
        conn.close()
        
        logger.info("当前商品库存:")
        for name, stock, price in products:
            logger.info(f"  {name}: 库存 {stock}, 价格 ${price}")

# 演示批量事务回滚
def demo_batch_transaction():
    demo = BatchTransactionDemo()
    
    print("=== 初始库存 ===")
    demo.show_inventory()
    
    # 成功的批量订单
    print("\n=== 成功批量订单演示 ===")
    success_orders = [('Laptop', 2), ('Mouse', 5)]
    result = demo.create_order_batch(success_orders)
    print(f"批量订单结果: {'成功' if result else '失败'}")
    demo.show_inventory()
    
    # 失败的批量订单（会触发回滚）
    print("\n=== 失败批量订单演示（触发回滚）===")
    # 先恢复一些库存用于演示
    conn = sqlite3.connect(demo.db_path)
    cursor = conn.cursor()
    cursor.execute("UPDATE products SET stock = 10 WHERE name = 'Laptop'")
    conn.commit()
    conn.close()
    
    fail_orders = [('Laptop', 5), ('Keyboard', 100)]  # Keyboard库存只有30，会失败
    result = demo.create_order_batch(fail_orders)
    print(f"批量订单结果: {'成功' if result else '失败'}")
    
    print("\n=== 回滚后库存（验证数据一致性）===")
    demo.show_inventory()

if __name__ == "__main__":
    demo_batch_transaction()

3. 使用SQLAlchemy实现ORM事务回滚

SQLAlchemy是Python最流行的ORM库，提供了更高级的事务管理抽象。

案例3：SQLAlchemy ORM事务回滚

from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.exc import SQLAlchemyError
import datetime
import logging

Base = declarative_base()
logger = logging.getLogger(__name__)

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    balance = Column(Float, default=0.0)
    
    # 关系
    orders = relationship("Order", back_populates="user")

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    stock = Column(Integer, default=0)
    price = Column(Float, nullable=False)
    
    # 关系
    orders = relationship("Order", back_populates="product")

class Order(Base):
    __tablename__ = 'orders'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    product_id = Column(Integer, ForeignKey('products.id'))
    quantity = Column(Integer, nullable=False)
    total_price = Column(Float, nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    
    # 关系
    user = relationship("User", back_populates="orders")
    product = relationship("Product", back_populates="orders")

class SQLAlchemyTransactionDemo:
    def __init__(self, database_url='sqlite:///:memory:'):
        self.engine = create_engine(database_url, echo=True)  # echo=True 显示SQL语句
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.Session = Session
        self.init_data()
    
    def init_data(self):
        """初始化测试数据"""
        session = self.Session()
        try:
            # 清空现有数据
            session.query(Order).delete()
            session.query(User).delete()
            session.query(Product).delete()
            
            # 添加测试用户
            users = [
                User(name='Alice', email='alice@example.com', balance=1000.0),
                User(name='Bob', email='bob@example.com', balance=500.0)
            ]
            
            # 添加测试商品
            products = [
                Product(name='iPhone', stock=10, price=999.99),
                Product(name='AirPods', stock=50, price=199.99),
                Product(name='iPad', stock=20, price=599.99)
            ]
            
            session.add_all(users + products)
            session.commit()
            logger.info("SQLAlchemy 测试数据初始化完成")
            
        except Exception as e:
            session.rollback()
            logger.error(f"初始化数据失败: {e}")
        finally:
            session.close()
    
    def place_order_with_session_rollback(self, user_name: str, product_name: str, quantity: int):
        """
        使用session级别的回滚
        """
        session = self.Session()
        try:
            logger.info(f"开始下单: 用户={user_name}, 商品={product_name}, 数量={quantity}")
            
            # 查询用户和商品
            user = session.query(User).filter_by(name=user_name).first()
            product = session.query(Product).filter_by(name=product_name).first()
            
            if not user:
                raise ValueError(f"用户不存在: {user_name}")
            if not product:
                raise ValueError(f"商品不存在: {product_name}")
            
            # 检查库存
            if product.stock < quantity:
                raise ValueError(f"库存不足: {product_name} (需要: {quantity}, 可用: {product.stock})")
            
            # 计算总价
            total_price = product.price * quantity
            
            # 检查用户余额
            if user.balance < total_price:
                raise ValueError(f"余额不足: {user_name} (需要: {total_price}, 可用: {user.balance})")
            
            # 执行订单操作
            # 1. 扣减库存
            product.stock -= quantity
            
            # 2. 扣减余额
            user.balance -= total_price
            
            # 3. 创建订单
            order = Order(
                user_id=user.id,
                product_id=product.id,
                quantity=quantity,
                total_price=total_price
            )
            session.add(order)
            
            # 提交事务
            session.commit()
            logger.info("订单创建成功，事务已提交")
            
            return order
            
        except SQLAlchemyError as e:
            # SQLAlchemy异常自动回滚
            session.rollback()
            logger.error(f"订单创建失败，事务已回滚: {e}")
            raise
        except Exception as e:
            # 其他异常手动回滚
            session.rollback()
            logger.error(f"订单创建失败，事务已回滚: {e}")
            raise
        finally:
            session.close()
    
    def complex_business_operation(self):
        """
        复杂的业务操作，演示嵌套事务和保存点
        """
        session = self.Session()
        try:
            logger.info("开始复杂业务操作")
            
            # 创建一个保存点
            savepoint = session.begin_nested()
            
            try:
                # 第一个子操作
                logger.info("执行第一个子操作")
                # ... 一些数据库操作
                
                # 第二个子操作（可能会失败）
                logger.info("执行第二个子操作")
                # 模拟一个错误
                raise ValueError("模拟第二个子操作失败")
                
                # 如果没有错误，释放保存点
                savepoint.commit()
                
            except Exception as e:
                # 回滚到保存点，但外层事务可以继续
                savepoint.rollback()
                logger.warning(f"子操作失败，已回滚到保存点: {e}")
                # 继续执行其他操作...
            
            # 第三个子操作（在回滚保存点后继续执行）
            logger.info("执行第三个子操作")
            user = session.query(User).filter_by(name='Alice').first()
            user.name = 'Alice Smith'
            
            # 提交外层事务
            session.commit()
            logger.info("复杂业务操作完成")
            
        except Exception as e:
            session.rollback()
            logger.error(f"复杂业务操作失败: {e}")
            raise
        finally:
            session.close()

# 演示SQLAlchemy事务回滚
def demo_sqlalchemy_transaction():
    demo = SQLAlchemyTransactionDemo()
    
    print("=== 测试正常下单 ===")
    try:
        order = demo.place_order_with_session_rollback('Alice', 'iPhone', 1)
        print(f"订单创建成功: ID={order.id}")
    except Exception as e:
        print(f"下单失败: {e}")
    
    print("\n=== 测试库存不足回滚 ===")
    try:
        order = demo.place_order_with_session_rollback('Bob', 'iPhone', 100)  # Bob余额不足
    except Exception as e:
        print(f"下单失败: {e}")
    
    print("\n=== 测试复杂业务操作 ===")
    try:
        demo.complex_business_operation()
    except Exception as e:
        print(f"复杂操作失败: {e}")

if __name__ == "__main__":
    demo_sqlalchemy_transaction()

4. 使用Django ORM实现事务回滚

Django作为流行的Web框架，提供了完善的事务管理支持。

案例4：Django ORM事务回滚

# 在Django项目中，首先在models.py中定义模型
"""
# models.py
from django.db import models

class Customer(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    balance = models.DecimalField(max_digits=10, decimal_places=2, default=0)

class Item(models.Model):
    name = models.CharField(max_length=100)
    stock = models.IntegerField(default=0)
    price = models.DecimalField(max_digits=8, decimal_places=2)

class Purchase(models.Model):
    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    item = models.ForeignKey(Item, on_delete=models.CASCADE)
    quantity = models.IntegerField()
    total_price = models.DecimalField(max_digits=10, decimal_places=2)
    created_at = models.DateTimeField(auto_now_add=True)
"""

# 在views.py或utils.py中使用事务
from django.db import transaction, DatabaseError
from django.shortcuts import get_object_or_404
from .models import Customer, Item, Purchase
import logging

logger = logging.getLogger(__name__)

def purchase_item_django(customer_id: int, item_id: int, quantity: int) -> bool:
    """
    使用Django事务处理购买操作
    """
    try:
        # 使用transaction.atomic装饰器或上下文管理器
        with transaction.atomic():
            logger.info(f"开始购买: 客户={customer_id}, 商品={item_id}, 数量={quantity}")
            
            # 获取对象和加锁（select_for_update防止并发问题）
            customer = get_object_or_404(Customer.objects.select_for_update(), id=customer_id)
            item = get_object_or_404(Item.objects.select_for_update(), id=item_id)
            
            # 业务逻辑验证
            if item.stock < quantity:
                raise ValueError(f"库存不足: {item.name}")
            
            total_price = item.price * quantity
            if customer.balance < total_price:
                raise ValueError("余额不足")
            
            # 执行数据库操作
            item.stock -= quantity
            customer.balance -= total_price
            
            # 保存对象
            item.save()
            customer.save()
            
            # 创建购买记录
            Purchase.objects.create(
                customer=customer,
                item=item,
                quantity=quantity,
                total_price=total_price
            )
            
            logger.info("购买成功，事务将自动提交")
            return True
            
    except (ValueError, DatabaseError) as e:
        logger.error(f"购买失败，事务已回滚: {e}")
        return False
    except Exception as e:
        logger.error(f"购买过程中发生未知错误: {e}")
        return False

# 使用装饰器的方式
from django.db import transaction
from functools import wraps

def handle_transaction(func):
    """事务处理装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            with transaction.atomic():
                return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"事务执行失败: {e}")
            raise
    return wrapper

@handle_transaction
def batch_process_orders(order_data_list):
    """批量处理订单的装饰器版本"""
    for order_data in order_data_list:
        # 处理每个订单的逻辑
        pass
    # 如果任何一步失败，整个函数都会回滚

5. 高级主题：分布式事务与补偿机制

在实际的微服务架构中，经常需要处理跨多个数据库或服务的事务，这时需要使用分布式事务模式。

案例5：基于消息队列的最终一致性方案

import pika
import json
import logging
from typing import Dict, Any
import time

logger = logging.getLogger(__name__)

class DistributedTransactionManager:
    def __init__(self, rabbitmq_host='localhost'):
        self.rabbitmq_host = rabbitmq_host
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=rabbitmq_host)
        )
        self.channel = self.connection.channel()
        
        # 声明队列
        self.channel.queue_declare(queue='order_events', durable=True)
        self.channel.queue_declare(queue='payment_events', durable=True)
        self.channel.queue_declare(queue='compensation_events', durable=True)
    
    def publish_event(self, queue: str, event_data: Dict[str, Any]):
        """发布事件"""
        self.channel.basic_publish(
            exchange='',
            routing_key=queue,
            body=json.dumps(event_data),
            properties=pika.BasicProperties(
                delivery_mode=2,  # 使消息持久化
            )
        )
        logger.info(f"事件已发布到 {queue}: {event_data}")
    
    def process_order_with_compensation(self, order_data: Dict[str, Any]):
        """
        处理订单并实现补偿机制
        当某个步骤失败时，执行补偿操作
        """
        transaction_id = order_data.get('transaction_id')
        compensation_actions = []  # 记录需要补偿的操作
        
        try:
            # 步骤1：创建订单
            logger.info(f"步骤1: 创建订单 {transaction_id}")
            self.publish_event('order_events', {
                'transaction_id': transaction_id,
                'action': 'create_order',
                'data': order_data,
                'compensation_action': 'cancel_order'
            })
            compensation_actions.append('cancel_order')
            
            # 步骤2：扣减库存
            logger.info(f"步骤2: 扣减库存 {transaction_id}")
            self.publish_event('order_events', {
                'transaction_id': transaction_id,
                'action': 'reduce_stock',
                'data': order_data,
                'compensation_action': 'restore_stock'
            })
            compensation_actions.append('restore_stock')
            
            # 步骤3：处理支付
            logger.info(f"步骤3: 处理支付 {transaction_id}")
            self.publish_event('payment_events', {
                'transaction_id': transaction_id,
                'action': 'process_payment',
                'data': order_data,
                'compensation_action': 'refund_payment'
            })
            compensation_actions.append('refund_payment')
            
            logger.info(f"订单 {transaction_id} 处理成功")
            
        except Exception as e:
            logger.error(f"订单 {transaction_id} 处理失败: {e}")
            # 执行补偿操作
            self.compensate_transaction(transaction_id, compensation_actions)
            raise
    
    def compensate_transaction(self, transaction_id: str, compensation_actions: list):
        """执行补偿操作"""
        logger.info(f"开始补偿事务 {transaction_id}, 补偿动作: {compensation_actions}")
        
        # 反向执行补偿动作
        for action in reversed(compensation_actions):
            self.publish_event('compensation_events', {
                'transaction_id': transaction_id,
                'action': action,
                'reason': 'previous_step_failed'
            })
            time.sleep(0.1)  # 模拟补偿操作间隔
        
        logger.info(f"事务 {transaction_id} 补偿完成")

# 使用示例
def demo_distributed_transaction():
    dtm = DistributedTransactionManager()
    
    order_data = {
        'transaction_id': 'ORDER_001',
        'customer_id': 123,
        'items': [
            {'product_id': 1, 'quantity': 2},
            {'product_id': 2, 'quantity': 1}
        ],
        'total_amount': 299.97
    }
    
    try:
        dtm.process_order_with_compensation(order_data)
        print("分布式事务执行成功")
    except Exception as e:
        print(f"分布式事务执行失败: {e}")

if __name__ == "__main__":
    demo_distributed_transaction()