> ## 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.

# 信号机制
- URL: https://blog.vercanti.com/xin-hao-ji-zhi/
- Published: 2026-08-28T14:34:52.000Z
- Updated: 2026-08-28T14:57:33.000Z
- Description: Tortoise ORM 提供 4 种内置信号，覆盖模型的保存和删除生命周期： 不同信号的 handler 签名略有差异： 所有信号 handler 必须是 async 函数，可以在其中执行任意异步操作： 信号 handler 必须在应用启动时完成注册，推荐在 lifespan 中或模块顶层导入时注册。 将所有信号 handler 放在独立文件，在应用启动前导入： 场景：在 post_save handler 中访问 instance.author（ForeignKey 字段）。 现象：NoValuesFetched: author 或返回空值。 原因：
- Author: yellowdog
- Tags: Tortoise-orm

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

---

## 一、信号类型

Tortoise ORM 提供 4 种内置信号，覆盖模型的保存和删除生命周期：

| 信号           | 触发时机                                 |
| ------------ | ------------------------------------ |
| pre\_save    | Model.save() 执行前（包括 create 和 update） |
| post\_save   | Model.save() 执行后（包括 create 和 update） |
| pre\_delete  | Model.delete() 执行前                   |
| post\_delete | Model.delete() 执行后                   |

```python
from tortoise.signals import pre_save, post_save, pre_delete, post_delete

```

> 注意：信号只在通过模型实例操作时触发（`instance.save()`、`instance.delete()`）。批量操作如 `Model.filter(...).update(...)` 和 `Model.filter(...).delete()` 不触发信号。

---

## 二、注册方式

### 2.1 装饰器方式（推荐）

```python
from tortoise.signals import post_save
from tortoise import BaseDBAsyncClient
from myapp.models import User

@post_save(User)
async def on_user_created(
    sender: type,
    instance: User,
    created: bool,
    using_db: BaseDBAsyncClient,
    update_fields: list[str],
) -> None:
    if created:
        # 新用户注册后发送欢迎邮件
        await send_welcome_email(instance.email)

```

### 2.2 `Signal.connect()` 方式

```python
from tortoise.signals import post_save
from myapp.models import User

async def on_user_saved(sender, instance, created, using_db, update_fields):
    if created:
        await send_welcome_email(instance.email)

# 注册
post_save.connect(on_user_saved, sender=User)

# 取消注册
post_save.disconnect(on_user_saved, sender=User)

```

#### `Signal.connect()` 参数

| 参数      | 类型                    | 默认值  | 说明                       |
| ------- | --------------------- | ---- | ------------------------ |
| handler | Callable              | 无，必填 | 信号处理函数（必须是 async 函数）     |
| sender  | Type\[Model\] \| None | None | 限定触发信号的模型类；None 表示响应所有模型 |

#### `Signal.disconnect()` 参数

| 参数      | 类型                    | 默认值  | 说明                              |
| ------- | --------------------- | ---- | ------------------------------- |
| handler | Callable              | 无，必填 | 要取消注册的处理函数，必须与 connect 时传入的对象相同 |
| sender  | Type\[Model\] \| None | None | 与 connect 时保持一致                 |

---

## 三、Handler 函数签名

不同信号的 handler 签名略有差异：

### 3.1 `pre_save` / `post_save`

```python
async def handler(
    sender: type,               # 发出信号的模型类（如 User）
    instance: Model,            # 触发信号的模型实例
    created: bool,              # True 表示新创建，False 表示更新
    using_db: BaseDBAsyncClient,  # 当前使用的数据库连接
    update_fields: list[str],   # 本次 save() 指定更新的字段列表；若为空列表表示未指定（全量保存）
) -> None:
    ...

```

| 参数             | 类型                | 说明                                              |
| -------------- | ----------------- | ----------------------------------------------- |
| sender         | type              | 发出信号的模型类本身（不是实例）                                |
| instance       | Model             | 当前操作的模型实例，可读取其字段值                               |
| created        | bool              | True \= INSERT 操作，False \= UPDATE 操作            |
| using\_db      | BaseDBAsyncClient | 当前事务或连接上下文，可用于在同一事务内执行额外查询                      |
| update\_fields | list\[str\]       | save(update\_fields=\[...\]) 时传入的字段列表；全量保存时为空列表 |

### 3.2 `pre_delete` / `post_delete`

```python
async def handler(
    sender: type,               # 发出信号的模型类
    instance: Model,            # 触发信号的模型实例
    using_db: BaseDBAsyncClient,  # 当前使用的数据库连接
) -> None:
    ...

```

| 参数        | 类型                | 说明              |
| --------- | ----------------- | --------------- |
| sender    | type              | 发出信号的模型类        |
| instance  | Model             | 即将被删除或已被删除的模型实例 |
| using\_db | BaseDBAsyncClient | 当前连接上下文         |

---

## 四、异步 Handler 写法

所有信号 handler 必须是 `async` 函数，可以在其中执行任意异步操作：

```python
from tortoise.signals import post_save, pre_delete
from tortoise import BaseDBAsyncClient
from myapp.models import User, AuditLog

@post_save(User)
async def audit_user_change(
    sender: type,
    instance: User,
    created: bool,
    using_db: BaseDBAsyncClient,
    update_fields: list[str],
) -> None:
    action = "created" if created else "updated"
    await AuditLog.create(
        model="User",
        object_id=instance.pk,
        action=action,
        fields=",".join(update_fields) if update_fields else "all",
    )

@pre_delete(User)
async def audit_user_deletion(
    sender: type,
    instance: User,
    using_db: BaseDBAsyncClient,
) -> None:
    await AuditLog.create(
        model="User",
        object_id=instance.pk,
        action="deleted",
        fields="",
    )

```

---

## 五、常见应用场景

### 5.1 创建后发送通知

```python
from tortoise.signals import post_save
from myapp.models import Order

@post_save(Order)
async def notify_on_order_created(
    sender: type,
    instance: Order,
    created: bool,
    using_db,
    update_fields,
) -> None:
    if not created:
        return
    # 发送通知给用户
    await push_notification(
        user_id=instance.user_id,
        title="订单已创建",
        body=f"订单 #{instance.id} 已成功提交"
    )

```

### 5.2 删除前清理关联资源

```python
from tortoise.signals import pre_delete
from myapp.models import Article
import os

@pre_delete(Article)
async def cleanup_article_files(
    sender: type,
    instance: Article,
    using_db,
) -> None:
    # 删除文章关联的图片文件
    if instance.cover_image_path:
        try:
            os.remove(instance.cover_image_path)
        except FileNotFoundError:
            pass

    # 删除关联的评论（如未设置级联删除）
    await instance.fetch_related("comments")
    for comment in instance.comments:
        await comment.delete()

```

### 5.3 审计日志

```python
from tortoise.signals import post_save, post_delete
from myapp.models import User, AuditLog
import json

@post_save(User)
async def log_user_change(sender, instance, created, using_db, update_fields):
    await AuditLog.create(
        table_name="user",
        record_id=instance.pk,
        action="INSERT" if created else "UPDATE",
        changed_fields=json.dumps(update_fields),
        snapshot=json.dumps({
            "username": instance.username,
            "email": instance.email,
            "is_active": instance.is_active,
        }),
    )

@post_delete(User)
async def log_user_deletion(sender, instance, using_db):
    await AuditLog.create(
        table_name="user",
        record_id=instance.pk,
        action="DELETE",
        changed_fields="",
        snapshot="",
    )

```

### 5.4 密码自动哈希

```python
from tortoise.signals import pre_save
from passlib.hash import bcrypt
from myapp.models import User

@pre_save(User)
async def hash_password(sender, instance, created, using_db, update_fields):
    # 仅当 password_hash 字段有变更时才重新哈希
    if update_fields and "password_hash" not in update_fields:
        return
    raw = instance.password_hash
    # 如果已经是 bcrypt 格式则跳过
    if raw and not raw.startswith("$2b$"):
        instance.password_hash = bcrypt.hash(raw)

```

---

## 六、在 FastAPI 中注册信号的时机

信号 handler 必须在应用启动时完成注册，推荐在 `lifespan` 中或模块顶层导入时注册。

### 6.1 通过模块导入（推荐）

将所有信号 handler 放在独立文件，在应用启动前导入：

```python
# myapp/signals.py
from tortoise.signals import post_save, pre_delete
from myapp.models import User, Order

@post_save(User)
async def on_user_saved(sender, instance, created, using_db, update_fields):
    ...

@pre_delete(Order)
async def on_order_deleted(sender, instance, using_db):
    ...

```

```python
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from tortoise import Tortoise
import myapp.signals  # 导入即完成注册

@asynccontextmanager
async def lifespan(app: FastAPI):
    await Tortoise.init(config=TORTOISE_ORM)
    yield
    await Tortoise.close_connections()

app = FastAPI(lifespan=lifespan)

```

### 6.2 在 lifespan 中手动 connect

```python
from myapp.handlers import on_user_saved
from tortoise.signals import post_save
from myapp.models import User

@asynccontextmanager
async def lifespan(app: FastAPI):
    await Tortoise.init(config=TORTOISE_ORM)
    post_save.connect(on_user_saved, sender=User)  # 手动注册
    yield
    post_save.disconnect(on_user_saved, sender=User)  # 关闭时清理
    await Tortoise.close_connections()

```

---

## 七、踩坑与注意事项

### 7.1 信号中访问关联对象需显式 fetch

**场景**：在 `post_save` handler 中访问 `instance.author`（ForeignKey 字段）。

**现象**：`NoValuesFetched: author` 或返回空值。

**原因**：信号接收到的 `instance` 不自动预加载关联对象，需要显式 `fetch_related`。

```python
@post_save(Post)
async def on_post_saved(sender, instance, created, using_db, update_fields):
    # 错误：直接访问关联字段
    print(instance.author.email)  # NoValuesFetched

    # 正确：先 fetch_related
    await instance.fetch_related("author")
    print(instance.author.email)  # 正常

```

### 7.2 信号 handler 中抛异常的影响

信号 handler 中未被捕获的异常会向上传播，中断原始操作（对 `pre_save` 来说会阻止保存）。

```python
@pre_save(User)
async def validate_user(sender, instance, created, using_db, update_fields):
    if not instance.email.endswith("@company.com"):
        raise ValueError("只允许公司邮箱注册")  # 这会阻止 save() 完成

```

如果不希望 handler 中的错误影响主流程，需要自行捕获异常：

```python
@post_save(User)
async def send_email_safe(sender, instance, created, using_db, update_fields):
    if not created:
        return
    try:
        await send_welcome_email(instance.email)
    except Exception as e:
        # 记录日志，但不向上传播，避免影响用户创建流程
        logger.error(f"发送欢迎邮件失败: {e}")

```

### 7.3 批量操作不触发信号

`QuerySet.update()`、`QuerySet.delete()`、`Model.bulk_create()` 不会触发任何信号。

```python
# 不触发信号
await User.filter(is_active=False).delete()
await User.filter(id__in=[1, 2, 3]).update(is_active=True)

# 触发信号（逐条操作，性能较低）
users = await User.filter(is_active=False)
for user in users:
    await user.delete()  # 每次都触发 pre_delete 和 post_delete

```

需要在批量操作中执行额外逻辑时，应在业务代码层手动处理，而不是依赖信号。

### 7.4 同一信号注册多个 handler 的执行顺序

多个 handler 按注册顺序依次执行（串行，不并发）。后注册的 handler 晚执行。装饰器方式按代码加载顺序排列。

### 7.5 `pre_save` 中修改 instance 字段

在 `pre_save` handler 中修改 `instance` 的字段值，修改会反映到最终保存的 SQL 中：

```python
@pre_save(Article)
async def set_slug(sender, instance, created, using_db, update_fields):
    if created and not instance.slug:
        instance.slug = slugify(instance.title)  # 自动生成 slug，会被保存

```

---

## 八、最佳实践

**将所有 handler 集中到 `signals.py`，在应用启动时统一导入**：避免 handler 分散在各模块，保证注册顺序可预期，也便于排查"信号是否已注册"的问题。

```python
# myapp/signals.py — 所有 handler 在此定义
from tortoise.signals import post_save, pre_delete
from myapp.models import User, Order

@post_save(User)
async def on_user_created(sender, instance, created, using_db, update_fields):
    if created:
        await send_welcome_email(instance.email)

```

```python
# main.py
import myapp.signals  # 导入即注册，无需手动 connect

```

**handler 中的非关键副作用必须捕获异常**：发送邮件、推送通知等操作失败不应回滚主流程。在 handler 内部 try/except，记录日志后继续。

```python
@post_save(User)
async def send_notification(sender, instance, created, using_db, update_fields):
    if not created:
        return
    try:
        await push_notification(instance.id)
    except Exception as e:
        logger.error(f"通知发送失败: {e}")  # 不上抛，不影响用户创建

```

**在 `pre_save` 中校验业务规则，比在视图层校验更可靠**：无论通过哪个入口保存数据，`pre_save` 都会执行，可作为最后一道防线。抛出的异常会中断 `save()` 并向调用方传播。

**`post_save` 中需要关联数据时，使用 `using_db` 执行额外查询**：在同一连接/事务上下文中查询，避免读取到尚未提交的数据。

```python
@post_save(Order)
async def on_order_saved(sender, instance, created, using_db, update_fields):
    if created:
        user = await User.get(id=instance.user_id).using_db(using_db)
        await send_order_email(user.email, instance)

```

**避免在信号 handler 中修改触发信号的同一实例字段后再次 `save()`**：会导致无限递归（`pre_save` 触发 `save()`，`save()` 再触发 `pre_save`）。如需修改字段，在 `pre_save` 中直接修改 `instance` 字段值即可（会被一并写入 SQL），不需要再调用 `save()`。

---

## 九、常见陷阱

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

**现象：** 注册了 `post_delete` handler，但用 `User.filter(is_active=False).delete()` 删除后，handler 没有执行。

**原因：** 批量 `QuerySet.delete()` 和 `QuerySet.update()` 直接执行 SQL，不经过 Python 模型层，信号分发也在 Python 层，因此均不触发。

**解决：** 需要触发信号时，必须逐条操作。

```python
# 不触发信号
await User.filter(is_active=False).delete()

# 触发信号（性能较低）
users = await User.filter(is_active=False)
for user in users:
    await user.delete()

```

### 陷阱：handler 中访问关联字段抛出 `NoValuesFetched`

**现象：** 在 `post_save` handler 中访问 `instance.author` 或其他关联字段时，抛出 `NoValuesFetched` 异常。

**原因：** 信号接收到的 `instance` 是当前模型实例，不包含预加载的关联对象，和普通查询结果一样需要显式 fetch。

**解决：** 在 handler 中调用 `fetch_related` 或通过主键查询关联对象。

```python
@post_save(Post)
async def on_post_saved(sender, instance, created, using_db, update_fields):
    await instance.fetch_related('author')  # 必须先 fetch
    print(instance.author.email)

```

### 陷阱：在 `post_save` 中调用 `instance.save()` 导致无限递归

**现象：** 应用启动后保存某个模型时栈溢出，日志显示 `pre_save` / `post_save` 被无限次调用。

**原因：** 在 `post_save` handler 中调用了 `await instance.save()`，触发新的保存周期，进而再次触发 `post_save`，形成死循环。

**解决：** 在 `pre_save` 中直接修改 `instance` 字段（无需再次 save），或用条件标志位防止重入，或改用 `Model.filter(...).update(...)` 绕过信号。

```python
@pre_save(Article)
async def auto_fill(sender, instance, created, using_db, update_fields):
    if created and not instance.slug:
        instance.slug = slugify(instance.title)  # 直接赋值，不调用 save()

```

---

## 最佳实践

**信号处理器保持轻量**：`pre_save`/`post_save` 中执行耗时操作（发 HTTP 请求、写文件）会延长整个保存事务的时间；耗时操作放入后台队列（Celery），信号只触发任务。

**用 `pre_save` 做字段规范化，用 `post_save` 做级联操作**：`pre_save` 修改即将保存的字段值最合适；`post_save` 做通知、同步其他模型等依赖已保存数据的操作。

**避免在信号中嵌套调用触发同一信号的操作**：`post_save` 中调用 `await instance.save()` 会再次触发 `post_save`，形成无限递归。必须操作时用 `update_fields` 限定更新字段或临时取消注册。

**`Signals.pre_delete` 做软删除保护**：在 `pre_delete` 中检查业务约束（如仍有关联的订单），不满足时 raise 异常，比在 controller 层检查更可靠（避免直接 DB 操作绕过检查）。

**按模块注册信号，避免注册时机问题**：信号注册要在 ORM 初始化（`init_tortoise`/`register_tortoise`）之后执行，推荐在应用启动的 lifespan 中集中注册。

---

## 常见陷阱

### 陷阱：批量操作（`bulk_create`/`filter().update()`）不触发信号

**现象：** 期望 `post_save` 被调用，但批量创建/更新后信号处理器没有执行。  
**原因：** Tortoise ORM 的批量操作直接执行 SQL，不经过 ORM 对象生命周期，因此不触发 `pre_save`/`post_save`。  
**解决：** 批量操作后手动执行需要的副作用逻辑，或改用循环 `await obj.save()`（性能不如批量，需权衡）。

### 陷阱：`post_save` 中 `instance` 的关联字段未加载

**现象：** `post_save` 中访问 `instance.author`（外键关联）时报 `NoValuesFetched`。  
**原因：** `post_save` 接收的 `instance` 是刚保存的对象，关联字段未预加载，直接访问触发懒加载但 Tortoise 异步环境下无法同步执行。  
**解决：** 在信号处理器中显式 `await instance.fetch_related('author')` 后再访问：

```python
async def handler(sender, instance, **kwargs):
    await instance.fetch_related('author')
    print(instance.author.name)

```

### 陷阱：`pre_save` 中修改 instance 字段后仍需 save 才持久化

**现象：** 在 `pre_save` 中 `instance.slug = slugify(instance.title)` 修改字段，以为会自动持久化，但数据库没有变化。  
**原因：** `pre_save` 中修改的字段值会被纳入即将执行的 INSERT/UPDATE SQL 中，前提是修改的字段在原始更新操作的范围内；若是 `update_fields` 限定了字段，新增的 `slug` 不在列表中则不保存。  
**解决：** 确认 `update_fields` 参数包含了被修改的字段，或不使用 `update_fields`。

---

## 参见

- [模型字段完全参考](https://blog.vercanti.com/tortoise-orm-mo-xing-zi-duan-wan-quan-can-kao/)
- [事务与并发](https://blog.vercanti.com/tortoise-orm-shi-wu-yu-bing-fa/)
- [查询操作完全指南](https://blog.vercanti.com/tortoise-orm-cha-xun-cao-zuo-wan-quan-zhi-nan/)
- [初始化与配置](https://blog.vercanti.com/tortoise-orm-chu-shi-hua-yu-pei-zhi/)