Python事务安全案例如何保障事务安全

wen python案例 2026-06-28 289

本文目录导读：

数据库事务（以SQLAlchemy为例）
Redis事务（使用Pipeline）
文件系统事务（使用临时文件）
分布式事务（两阶段提交示例）
Python对象状态事务（Memento模式）

在Python中保障事务安全（Transaction Safety）的核心目标是在数据操作过程中保证 原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation） 和 持久性（Durability），即ACID特性，以下是几个典型的案例和实现方案：

数据库事务（以SQLAlchemy为例）

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)
session = Session()
try:
    # 开启事务
    session.begin()
    # 业务操作1：扣减账户A余额
    account_a = session.query(Account).filter_by(id=1).with_for_update().one()
    account_a.balance -= 100
    # 业务操作2：增加账户B余额
    account_b = session.query(Account).filter_by(id=2).with_for_update().one()
    account_b.balance += 100
    # 提交事务
    session.commit()
except Exception as e:
    # 回滚事务
    session.rollback()
    raise e
finally:
    session.close()

关键点：

with_for_update() 添加行级锁防止并发冲突
try/except 确保异常时自动回滚
commit() 确保所有操作原子性提交

Redis事务（使用Pipeline）

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def transfer_money(from_account, to_account, amount):
    pipe = r.pipeline(transaction=True)
    try:
        # 监控关键值
        pipe.watch(from_account, to_account)
        # 获取当前余额
        from_balance = int(pipe.get(from_account) or 0)
        to_balance = int(pipe.get(to_account) or 0)
        if from_balance < amount:
            raise ValueError("Insufficient funds")
        # 开始事务
        pipe.multi()
        pipe.decrby(from_account, amount)
        pipe.incrby(to_account, amount)
        pipe.execute()
        print("Transaction successful")
    except redis.WatchError:
        # 如果监控的值被修改，重试
        transfer_money(from_account, to_account, amount)
    finally:
        pipe.reset()

关键点：

WATCH 实现乐观锁
MULTI/EXEC 保证命令的原子性执行
WatchError 处理并发冲突重试

文件系统事务（使用临时文件）

import os
import tempfile
import shutil
def safe_update_config(config_path, new_content):
    # 创建临时文件
    temp_fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(config_path))
    try:
        with os.fdopen(temp_fd, 'w') as temp_file:
            # 写入新内容
            temp_file.write(new_content)
            temp_file.flush()
            os.fsync(temp_file.fileno())  # 确保数据写入磁盘
        # 原子替换文件
        os.replace(temp_path, config_path)
    except Exception:
        # 清理临时文件
        if os.path.exists(temp_path):
            os.remove(temp_path)
        raise

关键点：

先写入临时文件
os.replace() 原子替换（POSIX系统保证）
异常时删除临时文件保持一致性

分布式事务（两阶段提交示例）

from contextlib import contextmanager
@contextmanager
def distributed_transaction(services):
    """简化版两阶段提交"""
    prepared = []
    try:
        # Phase 1: 准备阶段
        for service in services:
            if not service.prepare():
                raise Exception(f"Prepare failed for {service}")
            prepared.append(service)
        # Phase 2: 提交阶段
        for service in prepared:
            service.commit()
    except Exception:
        # 回滚所有已准备的服务
        for service in prepared:
            try:
                service.rollback()
            except Exception:
                # 记录回滚失败日志，需要人工干预
                log.error(f"Rollback failed for {service}")
        raise

关键点：

两阶段提交（2PC）确保分布式一致性
准备阶段预留资源
失败时回滚所有参与者

Python对象状态事务（Memento模式）

class TransactionalObject:
    def __init__(self):
        self._state = {}
        self._checkpoint = None
    def begin_transaction(self):
        # 保存当前状态的快照
        self._checkpoint = self._state.copy()
    def commit(self):
        # 清除checkpoint
        self._checkpoint = None
    def rollback(self):
        if self._checkpoint is not None:
            self._state = self._checkpoint
            self._checkpoint = None
    def update(self, key, value):
        self._state[key] = value
# 使用示例
obj = TransactionalObject()
obj.begin_transaction()
obj.update("name", "Alice")
obj.update("age", 30)
# 如果操作失败，可以回滚
if error_condition:
    obj.rollback()  # 状态恢复
else:
    obj.commit()    # 确认修改