Flask中SQLAlchemy回滚失效求助（CSV批量导入MySQL场景）

• 事务提前提交：UserReposity的create方法内部执行了db.session.commit()，这会直接提交当前事务。如果create过程中出现错误，部分数据可能已经被提交到数据库，后续在processBlock的except块中调用rollback()时，已经无法回滚已提交的操作。
• 事务边界混乱：事务的提交逻辑被放在了Repository层，而非整个数据处理块的外层，导致每个create操作都独立完成事务，无法实现整个chunk数据的原子性导入（要么全成功，要么全回滚）。
• 冗余的批量操作：循环内重复调用bulk_save_objects，既增加了数据库交互次数，也加剧了事务提交的碎片化问题。

1. 调整Repository层逻辑，移除内部提交 #

让Repository只负责将对象添加到session，不执行事务提交，把事务控制权交给上层的处理块：

from api.models.doctor import User
from api.utils.database import db

class UserReposity:
    def create(self, chunk):
        user_object_list = []
        for line in chunk:
            user = User(
                USER_FIRST_NAME=line['First Name'],
                USER_LAST_NAME=line['Last Name']
            )
            user_object_list.append(user)
        
        # 仅将对象批量加入session，不提交事务
        db.session.bulk_save_objects(user_object_list, return_defaults=True)
        return user_object_list

2. 上层统一管理事务的提交与回滚 #

在processBlock中集中处理事务的提交，确保所有数据库操作完成后再统一提交，出错时直接回滚整个事务：

def processBlock(self, chunk):
    try:
        user_repo = UserReposity()
        list_users = user_repo.create(chunk)
        
        # 其他Repository的操作也遵循相同逻辑：只加入session，不提交
        # other_repo = OtherRepository()
        # other_repo.create(chunk)
        
        # 所有数据操作完成后，统一提交事务
        db.session.commit()
    except Exception as e:
        # 触发异常时回滚整个事务
        db.session.rollback()
        # 可在此添加错误日志记录
        print(f"导入失败，已回滚：{str(e)}")

额外注意事项 #

• 确保所有参与导入的Repository操作都不执行commit，全部由上层processBlock统一控制。
• 对于20k行的CSV，当前的分块处理逻辑可以保留，每块作为一个独立事务，避免单次事务数据量过大导致性能问题，同时某块出错也不会影响已成功提交的块。
• 回滚后可以根据需求决定是否终止整个导入流程，或跳过当前错误块继续处理下一块。

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