SQLAlchemy如何仅回滚最后一次session.add操作

编写从User表查询指定用户、不存在则自动创建的函数时，遇到唯一约束冲突触发IntegrityError后，调用session.rollback()会回滚整个会话的所有未提交修改，无法满足同一会话多次调用函数、调用间隙无需提交的需求，初始实现代码如下：

def get_user_id(session, username):
    query = session.query(User).filter(User.username == username)
    try:
        user = query.one()
    except NoResultFound as e:
        user = User(username=username)
        try:
            session.add(user)
            session.flush()
        except IntegrityError as e:
            session.rollback() # 会回滚会话内所有未提交变更
            user = query.one()
    return user.id_user

使用SQLAlchemy的**保存点（nested transaction）**实现局部回滚，不要直接调用全局session.rollback()。
保存点是数据库层面支持的嵌套事务机制，回滚时只会撤销当前保存点创建之后的操作，不会影响外层会话已经存在的其他未提交修改，修改后的代码如下：

from sqlalchemy.exc import NoResultFound, IntegrityError

def get_user_id(session, username):
    user_query = session.query(User).filter(User.username == username)
    # 先执行一次查询，减少不必要的保存点创建开销
    try:
        return user_query.one().id_user
    except NoResultFound:
        pass

    # 创建嵌套事务保存点
    savepoint = session.begin_nested()
    new_user = User(username=username)
    try:
        session.add(new_user)
        savepoint.commit()
        return new_user.id_user
    except IntegrityError:
        # 仅回滚当前保存点范围内的新增操作，外层其他未提交修改完全不受影响
        savepoint.rollback()
        return user_query.one().id_user

补充说明：

• 所有支持事务的主流数据库（PostgreSQL、MySQL InnoDB、SQLite等）都原生支持保存点特性，无需额外修改数据库配置
• SQLAlchemy 2.0+版本中也可以使用with session.begin(nested=True):的上下文写法，逻辑完全一致
• 捕获IntegrityError后回滚到保存点，会话不会进入失效状态，后续可以正常执行其他操作，无需提前提交外层事务

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