title: 深入解析事务基础与原子操作原理
date: 2025/04/23 12:07:08
updated: 2025/04/23 12:07:08
author: cmdragon

excerpt:
事务是数据库操作的核心，需满足ACID特性：原子性、一致性、隔离性和持久性。事务隔离级别包括读未提交、读已提交、可重复读和串行化，适用于不同场景。嵌套事务通过上下文管理实现，支持回滚点以部分回滚。实战案例展示了订单系统中的事务处理，包括余额检查和支付失败回滚。常见错误如TransactionManagementError和死锁，需通过正确使用事务装饰器和重试机制解决。课后Quiz探讨了事务隔离和嵌套事务回滚的处理方法。运行环境配置包括依赖安装和FastAPI启动设置。

categories:

• 后端开发
• FastAPI

tags:

• 事务处理
• ACID特性
• 嵌套事务
• 回滚点
• 订单系统
• 事务隔离级别
• 数据库操作

扫描二维码关注或者微信搜一搜：编程智域 前端至全栈交流与成长

探索数千个预构建的 AI 应用，开启你的下一个伟大创意

1. 事务基础与原子操作原理

1.1 事务的ACID特性

在数据库操作中，事务需要满足ACID特性：

• 原子性（Atomicity）：操作要么全部成功，要么全部失败
• 一致性（Consistency）：保持数据库的完整性约束
• 隔离性（Isolation）：并发事务相互隔离
• 持久性（Durability）：提交后永久保存

# 示例：银行转账的原子操作
from fastapi import APIRouter, Depends
from tortoise.transactions import atomic

router = APIRouter()


@router.post("/transfer")
@atomic()  # 使用装饰器包裹事务范围
async def transfer_money(
        from_account: str,
        to_account: str,
        amount: float
):
    # 扣减转出账户
    from_acc = await Account.get(number=from_account)
    from_acc.balance -= amount
    await from_acc.save()

    # 增加转入账户
    to_acc = await Account.get(number=to_account)
    to_acc.balance += amount
    await to_acc.save()

    return {"message": "转账成功"}

1.2 事务隔离级别对比

2. 嵌套事务实现与回滚点

2.1 嵌套事务上下文管理

from tortoise.transactions import in_transaction


async def complex_operation():
    async with in_transaction() as conn1:  # 外层事务
        await Model1.create(...)

        try:
            async with in_transaction(connection=conn1) as conn2:  # 内层事务
                await Model2.create(...)
                await conn2.rollback()  # 仅回滚内层操作
        except Exception:
            pass

        await Model3.create(...)  # 外层事务继续执行

2.2 回滚点（Savepoint）使用

async def savepoint_demo():
    async with in_transaction() as conn:
        savepoint = await conn.savepoint()  # 创建回滚点

        try:
            await Model.create(...)
            if error_condition:
                await savepoint.rollback()  # 回滚到保存点
        except Exception:
            await savepoint.rollback()

3. 完整实战案例：订单系统

from fastapi import Depends
from tortoise.contrib.fastapi import register_tortoise
from tortoise.models import Model
from tortoise import fields


# 数据模型定义
class User(Model):
    id = fields.IntField(pk=True)
    balance = fields.DecimalField(max_digits=10, decimal_places=2)


class Order(Model):
    id = fields.UUIDField(pk=True)
    user = fields.ForeignKeyField('models.User')
    amount = fields.DecimalField(max_digits=10, decimal_places=2)
    status = fields.CharField(max_length=20)


# 事务处理服务
class OrderService:
    @staticmethod
    @atomic()
    async def create_order(user_id: int, amount: float):
        user = await User.get(id=user_id)

        # 检查余额
        if user.balance 

4. 常见报错解决方案

4.1 TransactionManagementError

错误现象：
TransactionManagementError: Transaction not found for current context

解决方法：

1. 检查事务装饰器的使用范围
2. 确保异步函数正确使用async/await
3. 验证数据库连接配置是否正确

4.2 死锁检测

错误日志：
Deadlock found when trying to get lock

处理方案：

from tortoise.exceptions import OperationalError
from fastapi import HTTPException


async def safe_transaction():
    try:
        async with in_transaction():
    # 数据库操作
    except OperationalError as e:
        if "Deadlock" in str(e):
            await asyncio.sleep(0.1)  # 随机延迟后重试
            return await safe_transaction()
        else:
            raise HTTPException(status_code=500, detail="数据库错误")

5. 课后Quiz

5.1 事务隔离问题

问题：在可重复读隔离级别下，如何处理余额校验时的并发修改？

答案解析：
使用SELECT FOR UPDATE锁定记录：

async def update_balance(user_id: int):
    async with in_transaction():
        user = await User.select_for_update().get(id=user_id)
        # 后续操作

5.2 嵌套事务回滚

问题：当外层事务捕获内层事务的异常时，如何保证部分提交？

正确答案：
使用保存点机制：

async def nested_transaction():
    async with in_transaction() as conn:
        savepoint = await conn.savepoint()
        try:
        # 内层操作
        except Exception:
            await savepoint.rollback()
        # 外层继续执行

6. 运行环境配置

安装依赖：

pip install fastapi tortoise-orm uvicorn pydantic

启动配置：

# main.py
from fastapi import FastAPI

app = FastAPI()

register_tortoise(
    app,
    db_url='sqlite://db.sqlite3',
    modules={'models': ['models']},
    generate_schemas=True,
    add_exception_handlers=True
)

启动命令：

uvicorn main:app --reload

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长，阅读完整的文章：深入解析事务基础与原子操作原理 | cmdragon's Blog

往期文章归档：

• 掌握Tortoise-ORM高级异步查询技巧 | cmdragon's Blog
• FastAPI与Tortoise-ORM实现关系型数据库关联 | cmdragon's Blog
• Tortoise-ORM与FastAPI集成：异步模型定义与实践 | cmdragon's Blog
• 异步编程与Tortoise-ORM框架 | cmdragon's Blog
• FastAPI数据库集成与事务管理 | cmdragon's Blog
• FastAPI与SQLAlchemy数据库集成 | cmdragon's Blog
• FastAPI与SQLAlchemy数据库集成与CRUD操作 | cmdragon's Blog
• FastAPI与SQLAlchemy同步数据库集成 | cmdragon's Blog
• SQLAlchemy 核心概念与同步引擎配置详解 | cmdragon's Blog
• FastAPI依赖注入性能优化策略 | cmdragon's Blog
• FastAPI安全认证中的依赖组合 | cmdragon's Blog
• FastAPI依赖注入系统及调试技巧 | cmdragon's Blog
• FastAPI依赖覆盖与测试环境模拟 | cmdragon's Blog
• FastAPI中的依赖注入与数据库事务管理 | cmdragon's Blog
• FastAPI依赖注入实践：工厂模式与实例复用的优化策略 | cmdragon's Blog
• FastAPI依赖注入：链式调用与多级参数传递 | cmdragon's Blog
• FastAPI依赖注入：从基础概念到应用 | cmdragon's Blog
• FastAPI中实现动态条件必填字段的实践 | cmdragon's Blog
• FastAPI中Pydantic异步分布式唯一性校验 | cmdragon's Blog
• 掌握FastAPI与Pydantic的跨字段验证技巧 | cmdragon's Blog
• FastAPI中的Pydantic密码验证机制与实现 | cmdragon's Blog
• 深入掌握FastAPI与OpenAPI规范的高级适配技巧 | cmdragon's Blog
• Pydantic字段元数据指南：从基础到企业级文档增强 | cmdragon's Blog
• Pydantic Schema生成指南：自定义JSON Schema | cmdragon's Blog
• Pydantic递归模型深度校验36计：从无限嵌套到亿级数据的优化法则 | cmdragon's Blog
• Pydantic异步校验器深：构建高并发验证系统 | cmdragon's Blog
• Pydantic根校验器：构建跨字段验证系统 | cmdragon's Blog
• Pydantic配置继承抽象基类模式 | cmdragon's Blog
• Pydantic多态模型：用鉴别器构建类型安全的API接口 | cmdragon's Blog
• FastAPI性能优化指南：参数解析与惰性加载 | cmdragon's Blog
• FastAPI依赖注入：参数共享与逻辑复用 | cmdragon's Blog
• FastAPI安全防护指南：构建坚不可摧的参数处理体系 | cmdragon's Blog
• FastAPI复杂查询终极指南：告别if-else的现代化过滤架构 | cmdragon's Blog
• FastAPI 核心机制：分页参数的实现与最佳实践 | cmdragon's Blog