信号机制

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

一、信号类型

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

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

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

二、注册方式

2.1 装饰器方式（推荐）

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() 方式

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() 参数

Signal.disconnect() 参数

三、Handler 函数签名

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

3.1 pre_save / post_save

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

3.2 pre_delete / post_delete

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

四、异步 Handler 写法

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

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 创建后发送通知

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 删除前清理关联资源

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 审计日志

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 密码自动哈希

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 放在独立文件，在应用启动前导入：

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

# 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

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。

@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 来说会阻止保存）。

@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 中的错误影响主流程，需要自行捕获异常：

@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() 不会触发任何信号。

# 不触发信号
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 中：

@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 分散在各模块，保证注册顺序可预期，也便于排查"信号是否已注册"的问题。

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

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

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

@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 执行额外查询：在同一连接/事务上下文中查询，避免读取到尚未提交的数据。

@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 层，因此均不触发。

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

# 不触发信号
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 或通过主键查询关联对象。

@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(...) 绕过信号。

@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') 后再访问：

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。

参见

• 模型字段完全参考
• 事务与并发
• 查询操作完全指南
• 初始化与配置