Django `transaction.atomic()` 完整使用指南

目录

1. #概述
2. #基本用法
3. #事务一致性保障机制
4. #破坏一致性的常见场景
5. #高级用法
6. #最佳实践
7. #诊断与调试
8. #附录

概述

transaction.atomic() 是 Django 提供的数据库事务管理工具，用于确保一系列数据库操作要么全部成功提交，要么全部回滚，维护数据的一致性。

基本用法

1. 作为上下文管理器

from django.db import transaction

def view_func(request):
    with transaction.atomic():
        # 事务内的操作
        obj1.save()
        obj2.save()

2. 作为装饰器

@transaction.atomic
def view_func(request):
    # 整个函数都在事务中执行
    obj1.save()
    obj2.save()

事务一致性保障机制

• 进入块时：创建保存点或开始新事务
• 成功退出：提交所有更改
• 异常退出：自动回滚所有更改
• 嵌套处理：内层块作为保存点处理

破坏一致性的常见场景

1. 异常处理不当

with transaction.atomic():
    try:
        risky_operation()
    except Exception:
        logger.error("Error occurred")  # 不重新抛出异常 → 破坏一致性

修复方案：

with transaction.atomic():
    try:
        risky_operation()
    except Exception as e:
        logger.error("Error occurred")
        raise  # 重新抛出异常

2. 长时间运行事务

with transaction.atomic():
    process_data()  # 耗时操作
    update_database()  # 可能因锁超时失败

3. 混合非数据库操作

with transaction.atomic():
    save_to_db()
    call_external_api()  # 网络操作
    save_to_db_again()   # 外部API失败可能导致不一致

4. 多数据库操作

with transaction.atomic():  # 只针对default数据库
    ModelA.objects.create(...)  # default
    ModelB.objects.using('other_db').create(...)  # 不受保护

5. 隔离级别冲突

# 事务1
with transaction.atomic():
    item = Item.objects.select_for_update().get(pk=1)
    time.sleep(10)

# 事务2 (并发执行)
with transaction.atomic():
    item = Item.objects.get(pk=1)  # 可能被阻塞或读取过期数据

高级用法

1. 手动保存点控制

with transaction.atomic():
    a.save()
    sid = transaction.savepoint()  # 创建保存点
    
    try:
        b.save()
    except Exception:
        transaction.savepoint_rollback(sid)  # 回滚到保存点
        # 可继续其他操作
    
    transaction.savepoint_commit(sid)  # 释放保存点

2. 指定数据库

with transaction.atomic(using='replica_db'):
    ReplicaModel.objects.create(...)

3. 嵌套事务控制

with transaction.atomic():  # 外层事务
    a.save()
    
    try:
        with transaction.atomic():  # 内层保存点
            b.save()
            raise ValueError
    except ValueError:
        pass  # 仅回滚内层
    
    c.save()  # 仍会提交

最佳实践

1. 保持事务简短：只包含必要的数据库操作
2. 明确异常处理：
   • 要么完全处理并确保一致性
   • 要么重新抛出异常
3. 避免混合操作：
   • 不在事务中执行网络/文件IO
   • 不在事务中执行耗时计算
4. 多数据库处理：

   with transaction.atomic(using='default'):
       with transaction.atomic(using='other_db'):
           # 跨数据库操作
   

5. 测试隔离级别：了解数据库的默认隔离级别
6. 监控长事务：设置数据库长事务警报

诊断与调试

1. 检查事务状态

from django.db import connection

print(f"In atomic block: {connection.in_atomic_block}")
print(f"Needs rollback: {connection.needs_rollback}")

2. 检查隔离级别(PostgreSQL)

with connection.cursor() as cursor:
    cursor.execute("SHOW transaction_isolation")
    print(cursor.fetchone())

3. 日志记录

LOGGING = {
    'loggers': {
        'django.db.backends': {
            'level': 'DEBUG',
            'handlers': ['console'],
        }
    }
}

附录

支持的数据库后端

事务隔离级别参考