求助：如何在SQLAlchemy中实现多级别事务传播控制

你遇到的核心问题是错误地在子事务中提前提交了会话，以及原有装饰器未区分事务传播逻辑，导致嵌套事务无法按预期回滚。SQLAlchemy的事务模型和Spring不同，需要通过会话管理、保存点（Savepoint）及独立会话来模拟Spring的传播级别。

1. 定义事务传播级别枚举 #

先明确对应需求的三种传播级别：

from enum import Enum

class Propagation(Enum):
    # 需求1/2：加入父事务，父/子失败则全量回滚
    REQUIRED = "required"
    # 需求3：独立事务，子失败不影响父提交
    REQUIRES_NEW = "requires_new"
    # 可选：嵌套事务，子回滚仅回滚到保存点，父可继续提交
    NESTED = "nested"

2. 实现带传播级别的事务装饰器 #

通过线程本地存储跟踪当前线程的事务状态，结合SQLAlchemy的会话/保存点机制实现传播逻辑：

import threading
from contextlib import contextmanager
from flask_sqlalchemy import SQLAlchemy

# 假设你的SQLAlchemy实例为db
db = SQLAlchemy()
thread_local = threading.local()

@contextmanager
def transaction(propagation=Propagation.REQUIRED):
    current_session = getattr(thread_local, "active_session", None)
    own_session = False
    savepoint = None

    if propagation == Propagation.REQUIRED:
        # 无活跃事务则新建，否则加入现有事务
        if not current_session or not current_session.is_active:
            current_session = db.session
            current_session.begin()
            thread_local.active_session = current_session
            own_session = True

    elif propagation == Propagation.REQUIRES_NEW:
        # 创建独立会话，完全隔离父事务
        old_session = current_session
        current_session = db.create_scoped_session()
        current_session.begin()
        thread_local.active_session = current_session
        own_session = True

        try:
            yield current_session
            current_session.commit()
        except Exception as e:
            current_session.rollback()
            raise e
        finally:
            current_session.remove()
            thread_local.active_session = old_session
        return

    elif propagation == Propagation.NESTED:
        # 有父事务则创建保存点，无则新建事务
        if current_session and current_session.is_active:
            savepoint = current_session.begin_nested()
        else:
            current_session = db.session
            current_session.begin()
            thread_local.active_session = current_session
            own_session = True

    try:
        yield current_session
        if own_session:
            current_session.commit()
        elif savepoint:
            # 提交保存点（不影响父事务）
            current_session.commit()
    except Exception as e:
        if own_session:
            current_session.rollback()
        elif savepoint:
            # 仅回滚到保存点
            current_session.rollback(savepoint)
        raise e
    finally:
        if own_session:
            thread_local.active_session = None
            current_session.remove()

def make_transaction(propagation=Propagation.REQUIRED):
    def decorator(func):
        def wrapper(*args, **kwargs):
            with transaction(propagation=propagation):
                return func(*args, **kwargs)
        return wrapper
    return decorator

3. 仓库层改造 #

保持仓库层仅做CRUD操作，不处理事务提交：

class BaseRepository:
    def __init__(self, model):
        self.model = model

    def save(self, entity):
        db.session.add(entity)

    # 其他CRUD方法同理，仅操作会话不提交

4. 三种需求场景示例 #

场景1：父事务失败，所有子事务回滚 #

使用Propagation.REQUIRED，子事务加入父事务上下文：

@make_transaction(Propagation.REQUIRED)
def save_user(user):
    user_repo.save(user)
    save_user_profile(user.profile)
    # 父事务抛出异常，所有操作回滚
    raise RuntimeError("父事务执行失败")

@make_transaction(Propagation.REQUIRED)
def save_user_profile(profile):
    profile_repo.save(profile)

场景2：子事务失败，父事务回滚 #

同样使用Propagation.REQUIRED，子事务异常向上传递触发父事务回滚：

@make_transaction(Propagation.REQUIRED)
def save_user(user):
    user_repo.save(user)
    # 子事务异常未被捕获，父事务回滚
    save_user_profile(user.profile)

@make_transaction(Propagation.REQUIRED)
def save_user_profile(profile):
    profile_repo.save(profile)
    raise RuntimeError("子事务执行失败")

场景3：子事务失败，父事务正常提交 #

使用Propagation.REQUIRES_NEW，子事务独立于父事务：

@make_transaction(Propagation.REQUIRED)
def save_user(user):
    user_repo.save(user)
    try:
        save_user_profile(user.profile)
    except RuntimeError:
        # 捕获子事务异常，父事务继续提交
        print("子资料保存失败，不影响用户主数据")
    return user

@make_transaction(Propagation.REQUIRES_NEW)
def save_user_profile(profile):
    profile_repo.save(profile)
    raise RuntimeError("子事务执行失败")

关键注意事项 #

• REQUIRES_NEW必须使用独立会话：SQLAlchemy单个会话只能有一个活跃事务，因此需要创建scoped_session实现完全隔离。
• 线程本地存储避免跨线程污染：确保每个线程的事务状态独立，不会互相干扰。
• 数据库支持保存点：NESTED依赖数据库的SAVEPOINT特性，MySQL、PostgreSQL等主流数据库均支持。

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