> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# Tortoise ORM 事务与并发
- URL: https://blog.vercanti.com/tortoise-orm-shi-wu-yu-bing-fa/
- Published: 2026-08-28T14:34:51.000Z
- Updated: 2026-08-28T14:57:31.000Z
- Description: 适合需要更精细控制的场景： 自动保存点（推荐）：@atomic() 和 in_transaction() 天然支持嵌套，内层块会自动创建数据库保存点（SAVEPOINT）。内层块异常只回滚到进入该块前的状态，不影响外层事务。 手动保存点（底层 API）： 防止并发读写同一行时产生竞态条件： select_for_update() 参数（部分数据库支持）： 不使用行锁时，用 F 表达式进行原子更新： 用版本号或时间戳实现，不使用数据库锁： 1. 事务尽量短：持有锁的时间越长，并发阻塞越严重。不要在事务内进行网络请求、大量计算。 2. @atomic() 不
- Author: yellowdog
- Tags: Tortoise-orm

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

---

## 一、事务基础

### `@atomic()` 装饰器

```python
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()` 上下文管理器

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

```python
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)` — 指定连接

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

```

---

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

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

```python
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）**：

```python
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`（行锁）

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

```python
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()` 参数（部分数据库支持）：**

| 参数           | 类型    | 默认    | 说明                                                 |
| ------------ | ----- | ----- | -------------------------------------------------- |
| nowait       | bool  | False | 无法立即获取锁时抛出异常而非等待                                   |
| skip\_locked | bool  | False | 跳过已锁定的行（用于任务队列场景）                                  |
| of           | tuple | ()    | 指定锁哪张表（JOIN 场景），如 of=('user',)                     |
| no\_key      | bool  | False | 使用较弱的行锁（FOR NO KEY UPDATE，PostgreSQL），允许插入引用该行的子记录 |

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

```

---

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

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

```python
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('库存不足或商品不存在')

```

---

## 五、乐观锁

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

```python
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('更新失败，请重试')

```

---

## 六、多数据库

### 配置多个连接

```python
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',
            }
        }
    }
)

```

### 路由到不同连接

```python
# 写操作用主库
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, ...)

```

---

## 七、连接池配置

```python
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 集成最佳实践

```python
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)

```

---

## 九、常见并发问题和解法

| 问题      | 场景             | 解法                            |
| ------- | -------------- | ----------------------------- |
| 读-改-写竞态 | 扣库存、余额转账       | select\_for\_update() \+ 事务   |
| 重复插入    | 重复注册、重复下单      | unique 约束 + 捕获 IntegrityError |
| 统计不准    | 并发 count() 后插入 | 原子 F() 表达式                    |
| 幻读      | 事务内统计结果变化      | 可重复读隔离级别（PG 默认）               |
| 死锁      | 多事务交叉锁定        | 固定加锁顺序、缩短事务时长                 |

---

## 十、事务注意事项

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 请求、消息队列推送等慢操作会阻塞其他并发，显著降低吞吐量。将非数据库副作用移到事务提交后执行。

```python
# 错误：事务内发 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()` \+ 事务**：使用数据库行锁避免超卖，比乐观锁重试更适合争用激烈的场景。

```python
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()` 在数据库层原子计算。

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

```

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

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

```python
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 时显式传入时间字段。

```python
# 显式传入时间
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()` 内。

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

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

```

---

## 参见

- [模型字段完全参考](https://blog.vercanti.com/tortoise-orm-mo-xing-zi-duan-wan-quan-can-kao/)
- [查询操作完全指南](https://blog.vercanti.com/tortoise-orm-cha-xun-cao-zuo-wan-quan-zhi-nan/)
- [信号机制](https://blog.vercanti.com/xin-hao-ji-zhi/)
- [初始化与配置](https://blog.vercanti.com/tortoise-orm-chu-shi-hua-yu-pei-zhi/)