Python：原子操作

Question

在有一组函数作为一个原子操作按顺序运行的场景中。如果其中任何一个由于某种原因失败，则操作必须回滚。

Python 有没有工具可以实现这个功能？

示例：

create a folder >> move some files >> send an email >> append data to a log file > etc.

Answer 1

当您有一系列步骤需要顺利协同工作或在出现问题时完全撤消时，上下文管理器会非常有用。

这是示例代码：

from contextlib import contextmanager

# Create the context manager
@contextmanager
def atomic_operation():
    try:
        create_folder()
        move_files()
        send_email()
        append_to_log()
        yield  # Operation successful, commit
    except Exception as e:
        rollback_folder_creation()
        rollback_file_movement()
        rollback_email_sending()
        rollback_log_appending()
        raise e

# Using the context manager
with atomic_operation():
    print("All tasks completed successfully")

当

try

块内发生异常时，上例中的程序通过执行与异常类型匹配的

except

块来处理该异常。然后，它继续运行每个任务的回滚函数，以便正确撤消先前完成的任务。如果在特定任务之后发生异常，则只会调用该点之前完成的任务的回滚函数。

例如，如果在

create_folder()

之后发生异常，则只会调用

rollback_folder_creation()

。如果在

move_files()

之后发生异常，

rollback_folder_creation()

和

rollback_file_movement()

都会被调用，以此类推。因此，它会根据异常发生的位置运行适当的回滚函数，而不是执行所有四个回滚函数。

某些操作的回滚操作可能非常容易做到：

def rollback_folder_creation():
    folder_path = "/path/to/created/folder"
    if os.path.exists(folder_path):
        os.rmdir(folder_path)  # Remove the folder
        print("Folder creation rolled back")

def rollback_file_movement():
    source_path = "/path/to/original/file"
    destination_path = "/path/to/moved/file"
    if os.path.exists(destination_path):
        shutil.move(destination_path, source_path)  # Move the file back
        print("File movement rolled back")

但是前邮可能会比较麻烦。可能有两种方法来处理电子邮件回滚：

如果发送邮件操作失败，会自动触发回滚操作，并发送回滚邮件通知收件人。

只有在整个原子操作成功完成后，您才能创建电子邮件草稿并发送电子邮件。

try:
    create_folder()
    move_files()
    email_draft.create_draft()
    append_to_log()
    email_draft.send_email()