事务装饰器

除了 AOP 中间件自动管理事务外，项目还提供了事务装饰器，用于特殊场景下的事务控制。

@transactional 装饰器

@transactional 装饰器用于需要显式控制事务的函数：

from src.middleware.transaction import transactional

@transactional
def complex_operation(data):
    """复杂操作（需要显式事务控制）"""
    # 所有数据库操作在同一个事务中
    user = User(**data)
    user.save()

    # 如果抛出异常，整个事务会回滚
    if not check_condition(user):
        raise BizError("条件不满足")

    # 其他操作
    # ...

    return user

装饰器实现

from functools import wraps
from src.core.database import db

def transactional(func):
    """事务装饰器：确保函数在事务中执行"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            db.session.commit()
            return result
        except Exception as e:
            db.session.rollback()
            raise
    return wrapper

使用场景

1. 定时任务：定时任务不经过 HTTP 请求，不触发中间件
2. 批量导入：大量数据导入需要显式控制事务边界
3. 跨模块操作：多个模块的操作需要在同一个事务中

@transactional_response 装饰器

@transactional_response 装饰器用于返回 R 对象的函数，根据响应 code 自动决定提交或回滚：

from src.middleware.transaction import transactional_response

@transactional_response
def add_position(data):
    """新增岗位"""
    position = Position(**data)
    position.save()

    # 返回成功响应时自动提交
    return R.ok(msg="新增成功")

    # 返回失败响应时自动回滚
    # return R.failed("新增失败")

装饰器实现

from functools import wraps
from src.core.database import db

def transactional_response(func):
    """事务响应装饰器：根据响应 code 自动提交或回滚"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            # 检查响应是否为 R 对象
            if hasattr(result, 'get_json'):
                body = result.get_json()
                if body and body.get('code') == 0:
                    db.session.commit()
                else:
                    db.session.rollback()
            else:
                # 非 R 对象，默认提交
                db.session.commit()
            return result
        except Exception as e:
            db.session.rollback()
            raise
    return wrapper

使用场景

1. Service 层方法：Service 层方法返回 R 对象
2. 需要精细控制：需要根据业务逻辑决定是否提交

TransactionContext 上下文管理器

TransactionContext 上下文管理器用于需要立即提交的场景：

from src.middleware.transaction import TransactionContext

def import_data(file):
    """导入数据（需要立即提交）"""
    with TransactionContext():
        # 在此上下文中的操作会立即提交
        for row in parse_excel(file):
            position = Position(**row)
            position.save()

    # 上下文结束后自动提交
    return R.ok(msg="导入成功")

实现原理

from contextlib import contextmanager
from src.core.database import db

@contextmanager
def TransactionContext():
    """事务上下文管理器：立即提交"""
    try:
        yield
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        raise

使用场景

1. 批量导入：大量数据导入时，每批次立即提交
2. 分步操作：需要分步执行，每步成功后立即提交
3. 外部调用：调用外部服务前需要确保数据已持久化

装饰器选择指南

场景	推荐方式	说明
标准 CRUD	AOP 中间件	自动管理，无需额外代码
定时任务	@transactional	不经过 HTTP 请求
批量导入	TransactionContext	需要立即提交
Service 层	@transactional_response	根据响应 code 决定
复杂业务	@transactional	需要显式控制

注意事项

重要提示

1. 不要混用多种方式：同一函数不要同时使用 AOP 中间件和装饰器
2. 装饰器顺序：如果同时使用多个装饰器，事务装饰器应该在最外层
3. 异常处理：装饰器会自动捕获异常并回滚，但业务代码仍应处理特定异常
4. 嵌套事务：SQLAlchemy 不支持真正的嵌套事务，嵌套调用会使用同一个事务

总结

事务装饰器提供了灵活的事务控制方式，适用于特殊场景。@transactional 用于需要显式控制事务的函数，@transactional_response 用于返回 R 对象的函数，TransactionContext 用于需要立即提交的场景。选择合适的事务管理方式，可以简化业务代码，提高代码可维护性。