Tortoise ORM 事务与并发

官方文档：https://tortoise.github.io/transactions.html
适用版本：tortoise-orm >= 0.20（2026-05-07 核实）

一、事务基础

@atomic() 装饰器

from tortoise.transactions import atomic

@atomic()
async def transfer(from_id: int, to_id: int, amount: float):
    sender = await Account.select_for_update().get(id=from_id)
    receiver = await Account.select_for_update().get(id=to_id)

    if sender.balance < amount:
        raise ValueError('余额不足')

    await Account.filter(id=from_id).update(balance=F('balance') - amount)
    await Account.filter(id=to_id).update(balance=F('balance') + amount)
    # 函数正常返回 → 自动 COMMIT
    # 抛出任何异常 → 自动 ROLLBACK

in_transaction() 上下文管理器

适合需要更精细控制的场景：

from tortoise.transactions import in_transaction

async def create_order_with_items(order_data, items_data):
    async with in_transaction() as conn:
        order = await Order.create(using_db=conn, **order_data)
        for item in items_data:
            await OrderItem.create(using_db=conn, order=order, **item)
        # with 块正常结束 → COMMIT
        # 异常 → ROLLBACK

in_transaction(connection_name) — 指定连接

async with in_transaction('secondary') as conn:
    await SomeModel.create(using_db=conn, ...)

二、保存点（Savepoint）与嵌套事务

自动保存点（推荐）：@atomic() 和 in_transaction() 天然支持嵌套，内层块会自动创建数据库保存点（SAVEPOINT）。内层块异常只回滚到进入该块前的状态，不影响外层事务。

from tortoise.transactions import in_transaction, atomic

async with in_transaction() as conn:
    await User.create(using_db=conn, name='Alice')

    # 内层块自动创建 SAVEPOINT
    try:
        async with in_transaction() as conn2:
            await User.create(using_db=conn2, name='Bob', email='invalid')
            # 内层异常 → 回滚到 SAVEPOINT，Alice 不受影响
    except Exception:
        pass

    # Alice 的记录仍然会提交

手动保存点（底层 API）：

async with in_transaction() as conn:
    await User.create(using_db=conn, name='Alice')

    try:
        await conn.execute_script('SAVEPOINT sp1')
        await User.create(using_db=conn, name='Bob', email='invalid')
    except Exception:
        await conn.execute_script('ROLLBACK TO SAVEPOINT sp1')

优先使用嵌套 async with in_transaction() 自动保存点，更简洁且不易出错。

三、SELECT FOR UPDATE（行锁）

防止并发读写同一行时产生竞态条件：

from tortoise.expressions import F

async with in_transaction() as conn:
    # 锁定该行，其他事务需要等待锁释放
    product = await Product.select_for_update().get(id=product_id)

    if product.stock < quantity:
        raise ValueError('库存不足')

    await Product.filter(id=product_id).update(
        stock=F('stock') - quantity,
        using_db=conn,
    )

select_for_update() 必须在事务内使用，否则锁无意义。

select_for_update() 参数（部分数据库支持）：

# 任务队列场景：跳过已被其他 worker 锁定的任务
tasks = await Task.filter(status='pending').select_for_update(skip_locked=True).limit(10)

四、原子操作（避免竞态）

不使用行锁时，用 F 表达式进行原子更新：

from tortoise.expressions import F

# 原子自增（安全，不需要先 SELECT）
await Product.filter(id=1).update(view_count=F('view_count') + 1)

# 原子库存扣减（加 filter 保证不会扣为负数）
updated = await Product.filter(
    id=product_id,
    stock__gte=quantity   # 乐观锁：只有库存足够时才更新
).update(stock=F('stock') - quantity)

if updated == 0:
    raise ValueError('库存不足或商品不存在')

五、乐观锁

用版本号或时间戳实现，不使用数据库锁：

class Product(Model):
    stock = fields.IntField()
    version = fields.IntField(default=0)   # 版本号

async def deduct_stock(product_id: int, quantity: int, max_retry: int = 3):
    for attempt in range(max_retry):
        product = await Product.get(id=product_id)

        if product.stock < quantity:
            raise ValueError('库存不足')

        # 只有版本号匹配时才更新（CAS 操作）
        updated = await Product.filter(
            id=product_id,
            version=product.version   # 乐观锁条件
        ).update(
            stock=F('stock') - quantity,
            version=F('version') + 1
        )

        if updated == 1:
            return True   # 成功

        # 版本号不匹配说明被其他请求修改，重试
        await asyncio.sleep(0.01 * (attempt + 1))

    raise RuntimeError('更新失败，请重试')

六、多数据库

配置多个连接

await Tortoise.init(
    config={
        'connections': {
            'default': 'postgres://user:pass@localhost/main_db',
            'replica': 'postgres://user:pass@replica/main_db',
        },
        'apps': {
            'models': {
                'models': ['myapp.models'],
                'default_connection': 'default',
            }
        }
    }
)

路由到不同连接

# 写操作用主库
await User.create(using_db='default', username='alice')

# 读操作用从库
users = await User.all().using_db('replica')

# 事务必须在同一连接内
async with in_transaction('default') as conn:
    user = await User.create(using_db=conn, ...)

七、连接池配置

await Tortoise.init(
    db_url='postgres://user:pass@localhost/db',
    modules={'models': ['myapp.models']},
    # 连接池配置通过 db_url 参数传入（不同驱动不同）
)

# asyncpg（PostgreSQL）通过 URL 参数
db_url = 'postgres://user:pass@localhost/db?minsize=5&maxsize=20'

# aiosqlite（SQLite）无连接池
db_url = 'sqlite://./db.sqlite3'

八、FastAPI 集成最佳实践

from contextlib import asynccontextmanager
from fastapi import FastAPI
from tortoise import Tortoise

TORTOISE_CONFIG = {
    'connections': {'default': 'postgres://user:pass@localhost/db'},
    'apps': {
        'models': {
            'models': ['myapp.models', 'aerich.models'],
            'default_connection': 'default',
        }
    },
    'timezone': 'Asia/Shanghai',
}

@asynccontextmanager
async def lifespan(app: FastAPI):
    await Tortoise.init(config=TORTOISE_CONFIG)
    await Tortoise.generate_schemas()   # 开发环境自动建表
    yield
    await Tortoise.close_connections()

app = FastAPI(lifespan=lifespan)

九、常见并发问题和解法

十、事务注意事项

1. 事务尽量短：持有锁的时间越长，并发阻塞越严重。不要在事务内进行网络请求、大量计算。
2. @atomic() 不可跨协程共享：事务连接绑定到当前执行上下文，不要把 conn 传给其他 task。
3. 异常后自动回滚：@atomic() 和 in_transaction() 中任何未捕获的异常都会触发 ROLLBACK。
4. auto_now 字段与事务：在事务中 update() 时，auto_now 不会触发；如需记录时间需显式传入。
5. SQLite 并发限制：SQLite 写操作会锁全表，不适合高并发写入场景，生产环境用 PostgreSQL/MySQL。
6. 连接泄漏：始终用上下文管理器（async with in_transaction()）而非手动 begin/commit，确保连接正确释放。

最佳实践

事务内只做数据库操作，不做网络请求：持有事务锁期间执行 HTTP 请求、消息队列推送等慢操作会阻塞其他并发，显著降低吞吐量。将非数据库副作用移到事务提交后执行。

# 错误：事务内发 HTTP 请求
@atomic()
async def create_order(data):
    order = await Order.create(**data)
    await send_email(order.user.email)  # 慢操作持有锁

# 正确：事务后执行副作用
@atomic()
async def create_order(data):
    order = await Order.create(**data)
    return order

order = await create_order(data)
await send_email(order.user.email)  # 事务已提交

高并发库存/余额场景用 select_for_update() + 事务：使用数据库行锁避免超卖，比乐观锁重试更适合争用激烈的场景。

async with in_transaction() as conn:
    product = await Product.select_for_update().get(id=pid)
    if product.stock < qty:
        raise ValueError('库存不足')
    await Product.filter(id=pid).update(stock=F('stock') - qty, using_db=conn)

无锁原子更新用 F() 表达式：计数器、访问量等只做增减操作的字段，不需要先 SELECT，直接用 F() 在数据库层原子计算。

await Article.filter(id=aid).update(view_count=F('view_count') + 1)

嵌套事务用嵌套 async with in_transaction()，不要手动 SAVEPOINT：内层块自动映射为数据库 SAVEPOINT，内层异常只回滚内层操作，比手动管理更安全。

多连接操作时明确指定 using_db：在事务上下文中创建/查询模型时，始终传入 using_db=conn，否则可能使用默认连接而脱离事务。

async with in_transaction() as conn:
    user = await User.create(using_db=conn, name='Alice')
    await Profile.create(using_db=conn, user=user)

常见陷阱

陷阱：@atomic() 跨协程共享事务连接

现象： 在 @atomic() 函数内用 asyncio.create_task() 或 await asyncio.gather() 启动子任务，子任务中的数据库操作不在同一事务中，或出现 "connection already closed" 错误。

原因： 事务连接绑定到当前执行上下文（通过上下文变量传递），create_task 创建的新 Task 继承的是上下文副本，对连接的修改不会回传，行为不确定。

解决： 不要在事务内并行调度子 Task。需要并行时，在事务外 gather，然后在事务内做最终写入。

陷阱：批量操作不触发信号和 auto_now

现象： 使用 Model.filter(...).update(...) 批量更新后，updated_at（auto_now=True）没有变化，注册的信号 handler 也没有执行。

原因： 批量 update() 和 delete() 直接执行 SQL，不经过 Python 模型层，auto_now 赋值和信号分发都在 Python 层，因此均不触发。

解决： 需要触发信号或 auto_now 时，必须逐条 instance.save() 或 instance.delete()；或在批量 update 时显式传入时间字段。

# 显式传入时间
from datetime import datetime, timezone
await Post.filter(status='draft').update(
    status='published',
    updated_at=datetime.now(timezone.utc),
)

陷阱：select_for_update() 在事务外使用无效

现象： 调用 select_for_update() 后，并发请求仍然读到相同值，锁没有生效。

原因： SELECT FOR UPDATE 必须在事务中才有意义，事务提交后锁自动释放；在事务外调用，PostgreSQL/MySQL 会忽略锁或立即释放。

解决： 始终将 select_for_update() 放在 async with in_transaction() 或 @atomic() 内。

# 错误：无事务，锁立即释放
product = await Product.select_for_update().get(id=pid)

# 正确
async with in_transaction() as conn:
    product = await Product.select_for_update().get(id=pid)

参见

• 模型字段完全参考
• 查询操作完全指南
• 信号机制
• 初始化与配置