Python实现原子函数：非数据库场景下如何添加事务支持？

Python本身确实没有针对数据库外场景的内置事务支持，要实现create_user_record和create_directory的原子性——即要么全部成功，要么全部回滚，核心思路是为每个操作配套对应的回滚逻辑，通过异常捕获控制执行流程：一旦某一步失败，就反向执行已完成操作的回滚步骤。

具体实现方案 #

1. 为每个操作定义回滚函数

   • 对应create_user_record，需要delete_user_record来删除已创建的用户记录
   • 对应create_directory，需要delete_directory来删除已创建的目录

2. 按顺序执行操作并捕获异常
   依次执行两个操作，每完成一个就记录下来；如果中途抛出异常，就反向遍历已完成的操作，逐个执行回滚。

示例代码 #

def create_user_record(user_record):
    # 实际创建用户记录的业务逻辑
    print(f"创建用户记录: {user_record}")
    # 可模拟失败场景：raise ValueError("用户记录创建失败")

def delete_user_record(user_record):
    # 回滚逻辑：删除已创建的用户记录
    print(f"回滚：删除用户记录: {user_record}")

def create_directory(directory_name):
    # 实际创建目录的业务逻辑
    print(f"创建目录: {directory_name}")
    # 可模拟失败场景：raise OSError("目录创建失败")

def delete_directory(directory_name):
    # 回滚逻辑：删除已创建的目录
    print(f"回滚：删除目录: {directory_name}")

def atomic_create_user_and_dir(user_record, directory_name):
    completed_ops = []  # 记录已成功执行的操作，用于回滚
    try:
        # 第一步：创建用户记录
        create_user_record(user_record)
        completed_ops.append(("user", user_record))
        
        # 第二步：创建目录
        create_directory(directory_name)
        completed_ops.append(("dir", directory_name))
        
        print("所有操作执行成功")
    except Exception as e:
        print(f"操作失败，开始回滚: {str(e)}")
        # 反向执行回滚，确保先回滚后执行的操作
        for op_type, op_data in reversed(completed_ops):
            if op_type == "user":
                delete_user_record(op_data)
            elif op_type == "dir":
                delete_directory(op_data)
        # 重新抛出异常，让调用方感知操作失败
        raise

关键注意事项 #

• 回滚函数的可靠性：回滚逻辑要尽量稳定，避免在回滚时抛出异常，必要时可给回滚代码加额外的异常捕获。
• 幂等性设计：回滚操作最好是幂等的（多次执行结果一致），比如删除不存在的用户记录/目录时，不抛出异常，防止回滚流程中断。
• 复杂场景扩展：如果需要添加更多操作，只需新增对应的执行和回滚函数，在completed_ops中记录即可，回滚逻辑无需大幅修改。

内容的提问来源于stack exchange，提问作者Yasin Amini