Django Signals 信號機制:觀察者模式實戰 | Django 教學
Django Signals(信號機制) 是 Django 內建的事件通知系統,實作了經典的 觀察者模式(Observer Pattern) 。當特定事件發生時——例如資料庫中的 Model 被儲存或刪除——Signal 能讓應用程式中的不同元件在 不直接依賴彼此 的情況下收到通知並執行對應動作。本篇將帶你從設計理念到實戰應用,全面掌握 Django Signals 的使用方式與最佳實踐。
Signal 設計模式:觀察者模式
觀察者模式(Observer Pattern) 是一種行為設計模式,定義了物件之間的一對多依賴關係。當一個物件(被觀察者)的狀態改變時,所有依賴它的物件(觀察者)都會自動收到通知。
Django Signal 的三個核心角色:
發送者(Sender) → 信號(Signal) → 接收者(Receiver)
User Model post_save create_user_profile()
Order Model post_save send_order_email()
Article Model post_delete cleanup_files()
- Signal(信號):事件的定義,例如
post_save代表「Model 儲存完成」事件 - Sender(發送者):觸發信號的物件,通常是 Model class
- Receiver(接收者):處理信號的 callable 函式,可以有多個接收者響應同一個信號
這種設計的最大好處是 解耦(Decoupling) ——發送者不需要知道有哪些接收者存在,接收者也不需要直接依賴發送者的程式碼。
內建信號一覽
Django 提供了多種內建信號,涵蓋 Model 操作、Request 處理和 Migration 等不同面向。
Model 信號
Model 信號是最常用的一類,當 Model 進行 CRUD 操作時觸發:
| 信號 | 觸發時機 | 常見用途 |
|---|---|---|
pre_save | save() 呼叫前 | 欄位預處理、自動計算值 |
post_save | save() 呼叫後 | 發送通知、建立關聯資料 |
pre_delete | delete() 呼叫前 | 備份資料、前置清理 |
post_delete | delete() 呼叫後 | 清理相關資源、刪除檔案 |
m2m_changed | ManyToMany 關聯變更時 | 同步相關資料、更新計數 |
pre_init | Model __init__ 前 | 初始化前攔截 |
post_init | Model __init__ 後 | 追蹤初始狀態 |
Request / Response 信號
| 信號 | 觸發時機 |
|---|---|
request_started | 每個 HTTP Request 開始處理時 |
request_finished | 每個 HTTP Request 處理完成時 |
got_request_exception | Request 處理過程中發生未捕獲的例外時 |
Management 指令信號
| 信號 | 觸發時機 |
|---|---|
pre_migrate | migrate 指令執行前 |
post_migrate | migrate 指令執行後 |
@receiver 裝飾器
@receiver 是連接 Signal 最推薦的方式,語法清晰且容易維護。
基本用法
# accounts/signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.conf import settings
from .models import UserProfile
@receiver(post_save, sender=settings.AUTH_USER_MODEL)
def create_user_profile(sender, instance, created, **kwargs):
"""新建 User 時自動建立 UserProfile"""
if created: # 只在新建時執行,更新時跳過
UserProfile.objects.create(user=instance)
接收函式的參數說明
每個接收函式都會收到一組固定參數,不同的 Signal 可能有額外的參數:
@receiver(post_save, sender=Article)
def on_article_saved(sender, instance, created, **kwargs):
"""
參數說明:
- sender:觸發此信號的 Model class(例如 Article)
- instance:觸發信號的物件實例(例如某一篇文章)
- created:True 表示新建,False 表示更新(僅 post_save 有此參數)
- **kwargs:必須接受,確保未來新增參數時的相容性
"""
if created:
print(f"新文章已建立:{instance.title}")
else:
print(f"文章已更新:{instance.title}")
post_save 實務範例
# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.mail import send_mail
from .models import Order
@receiver(post_save, sender=Order)
def notify_order_status(sender, instance, created, **kwargs):
"""訂單狀態變更時發送通知"""
if created:
# 新訂單建立
send_mail(
subject=f'訂單確認 #{instance.order_number}',
message=f'感謝您的訂購!訂單編號:{instance.order_number}',
from_email='noreply@example.com',
recipient_list=[instance.user.email],
fail_silently=True,
)
pre_save 實務範例
from django.db.models.signals import pre_save
from django.dispatch import receiver
from django.utils.text import slugify
from .models import Article
@receiver(pre_save, sender=Article)
def auto_generate_slug(sender, instance, **kwargs):
"""儲存文章前自動產生 slug"""
if not instance.slug:
instance.slug = slugify(instance.title, allow_unicode=True)
注意 :pre_save 中修改 instance 的欄位值會被自動帶入後續的 save() 操作,不需要額外呼叫 instance.save()(否則會造成無限遞迴)。
post_delete 實務範例
from django.db.models.signals import post_delete
from django.dispatch import receiver
import os
from .models import Article
@receiver(post_delete, sender=Article)
def cleanup_article_image(sender, instance, **kwargs):
"""刪除文章時清理封面圖片檔案"""
if instance.cover_image:
if os.path.isfile(instance.cover_image.path):
os.remove(instance.cover_image.path)
m2m_changed 實務範例
from django.db.models.signals import m2m_changed
from django.dispatch import receiver
from .models import Article
@receiver(m2m_changed, sender=Article.tags.through)
def on_article_tags_changed(sender, instance, action, pk_set, **kwargs):
"""文章標籤變更時更新標籤計數"""
# action 的值:pre_add, post_add, pre_remove, post_remove, pre_clear, post_clear
if action in ('post_add', 'post_remove', 'post_clear'):
instance.tag_count = instance.tags.count()
instance.save(update_fields=['tag_count'])
Signal.connect() 方法
除了 @receiver 裝飾器外,你也可以使用 Signal.connect() 方法來手動連接接收函式。這種方式更有彈性,支援動態連接與斷開。
from django.db.models.signals import post_save
from .models import Article
def on_article_saved(sender, instance, created, **kwargs):
"""文章儲存後的處理"""
if created:
print(f"新文章:{instance.title}")
# 手動連接 Signal
post_save.connect(on_article_saved, sender=Article)
# 手動斷開 Signal
post_save.disconnect(on_article_saved, sender=Article)
connect() 方法的完整簽名:
Signal.connect(
receiver, # 接收函式
sender=None, # 指定發送者(None 表示接收所有發送者的信號)
weak=True, # 是否使用弱引用(預設 True)
dispatch_uid=None # 唯一識別,防止重複連接
)
AppConfig.ready() 中註冊 Signal
Signal 的連接必須在 Django 的 App Registry(應用程式註冊表) 完全初始化後才能進行。官方推薦的做法是在 AppConfig.ready() 方法中匯入 signals 模組:
# accounts/apps.py
from django.apps import AppConfig
class AccountsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'accounts'
def ready(self):
# 匯入 signals 模組,觸發 @receiver 裝飾器的連接
import accounts.signals # noqa: F401
為什麼需要 ready()? 如果直接在 models.py 的頂層匯入 signals 模組,有可能在 Django 尚未完全啟動前就執行,導致 AppRegistryNotReady 錯誤。ready() 方法保證在所有 App 都已註冊完成後才會被呼叫。
檔案結構參考:
accounts/
├── __init__.py
├── apps.py # AppConfig.ready() 匯入 signals
├── models.py # Model 定義
├── signals.py # Signal 接收函式
├── admin.py
└── views.py
自訂 Signal
除了使用 Django 的內建信號外,你也可以定義自己的 自訂信號(Custom Signal) ,用來在應用程式中建立解耦的事件通知機制。
定義自訂 Signal
# orders/signals.py
from django.dispatch import Signal
# 定義自訂信號
order_completed = Signal() # 訂單完成
payment_received = Signal() # 收到付款
order_cancelled = Signal() # 訂單取消
發送自訂 Signal
# orders/models.py(或 services.py)
from django.db import models
from .signals import order_completed
class Order(models.Model):
order_number = models.CharField(max_length=20, unique=True)
status = models.CharField(max_length=20, default='pending')
user = models.ForeignKey('auth.User', on_delete=models.CASCADE)
total_amount = models.DecimalField(max_digits=10, decimal_places=2)
def complete(self):
"""完成訂單並發送信號"""
self.status = 'completed'
self.save()
# 發送自訂信號,附帶相關資料
order_completed.send(
sender=self.__class__,
order=self,
user=self.user,
)
接收自訂 Signal
# notifications/signals.py
from django.dispatch import receiver
from orders.signals import order_completed
@receiver(order_completed)
def send_completion_email(sender, order, user, **kwargs):
"""訂單完成時發送確認 Email"""
from django.core.mail import send_mail
send_mail(
subject=f'訂單已完成 #{order.order_number}',
message=f'您好 {user.username},您的訂單已完成!',
from_email='noreply@example.com',
recipient_list=[user.email],
fail_silently=True,
)
@receiver(order_completed)
def update_user_stats(sender, order, user, **kwargs):
"""訂單完成時更新使用者統計"""
user.profile.total_orders += 1
user.profile.total_spent += order.total_amount
user.profile.save()
注意:同一個 Signal 可以有多個接收者,它們會依序執行。
send() vs send_robust()
Django 提供兩種發送 Signal 的方式:
# send():若接收者拋出例外,會中斷後續接收者的執行
results = order_completed.send(sender=Order, order=order, user=user)
# send_robust():即使某個接收者拋出例外,仍會繼續執行其他接收者
results = order_completed.send_robust(sender=Order, order=order, user=user)
# results 是 [(receiver, response_or_exception), ...] 的 list
在生產環境中,建議使用 send_robust() 以避免單一接收者的錯誤影響其他接收者。
dispatch_uid 防止重複連接
在某些情境下(如測試環境或模組被多次匯入),Signal 接收函式可能會被重複連接,導致同一個事件觸發多次相同的處理。 dispatch_uid 參數可以解決這個問題:
from django.db.models.signals import post_save
from .models import Article
def on_article_saved(sender, instance, created, **kwargs):
print(f"文章已儲存:{instance.title}")
# 使用 dispatch_uid 確保只連接一次
post_save.connect(
on_article_saved,
sender=Article,
dispatch_uid='myapp.signals.on_article_saved' # 唯一識別字串
)
如果使用 @receiver 裝飾器搭配 AppConfig.ready() 的標準做法,通常不需要額外設定 dispatch_uid,因為 ready() 只會被呼叫一次。但在使用 connect() 方法且程式碼可能被多次執行的情境下,建議加上 dispatch_uid。
Signal 使用時機與反模式
了解何時該用 Signal、何時不該用,是避免架構混亂的關鍵。
適合使用 Signal 的情境
- 跨 App 解耦通知:App A 不應直接 import App B 的程式碼
- 通用行為追蹤:審計日誌(Audit Log)、活動記錄
- 事件觸發第三方整合:發送 Email、呼叫 Webhook
- 多個接收者響應同一事件:建立 Profile、發送通知、更新統計
Signal vs 覆寫 save() 的選擇指南
| 情境 | 建議方式 | 原因 |
|---|---|---|
| 同一 App 內的邏輯 | 覆寫 save() | 更直接、易讀、易測試 |
| 跨 App 通知 | Signal | 保持 App 間解耦 |
需要 created 判斷 | 兩者都可 | Signal 的 post_save 直接提供 |
| 多個地方需要響應同一事件 | Signal | Observer Pattern 天然支援多接收者 |
| 需要 Transaction 保證 | 覆寫 save() | Signal 在 Transaction 提交前觸發 |
反模式一:Signal 中放大量業務邏輯
# 錯誤:Signal 中夾雜複雜的業務邏輯,難以測試與維護
@receiver(post_save, sender=Order)
def process_order(sender, instance, created, **kwargs):
if created:
# 計算折扣
discount = calculate_discount(instance)
instance.discount = discount
# 更新庫存
for item in instance.items.all():
item.product.stock -= item.quantity
item.product.save()
# 通知物流
logistics_api.create_shipment(instance)
# 更新使用者積分
instance.user.points += instance.total * 0.1
instance.user.save()
# ...更多業務邏輯
# 正確:Signal 只做事件調度,業務邏輯放到 Service 層
@receiver(post_save, sender=Order)
def on_order_created(sender, instance, created, **kwargs):
if created:
OrderService.process_new_order(instance)
反模式二:連鎖 Signal(Signal 觸發 Signal)
# 危險:Article 的 post_save 觸發 User 的 save()
# 若 User 也有 post_save Signal,形成連鎖反應!
@receiver(post_save, sender=Article)
def update_author_stats(sender, instance, **kwargs):
instance.author.article_count = Article.objects.filter(
author=instance.author
).count()
instance.author.save() # 若 User 也有 post_save,可能引發連鎖!
解決方式:使用 update() 方法避免觸發 Signal,或使用 update_fields 限制更新的欄位:
# 改善方式一:使用 QuerySet.update()(不觸發 Signal)
from django.contrib.auth import get_user_model
User = get_user_model()
@receiver(post_save, sender=Article)
def update_author_stats(sender, instance, **kwargs):
count = Article.objects.filter(author=instance.author).count()
User.objects.filter(pk=instance.author.pk).update(article_count=count)
反模式三:忽略效能影響
Signal 是 同步執行 的。如果接收者中有耗時操作(如發送 Email、呼叫外部 API),會阻塞整個 Request 的處理。
# 不好的做法:Signal 中直接做耗時操作
@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
if created:
send_mail(...) # 阻塞 Request 直到 Email 發送完成
# 正確做法:耗時操作交給 Celery 非同步處理
@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
if created:
from .tasks import send_welcome_email_task
send_welcome_email_task.delay(instance.pk)
注意:QuerySet.update() 和 bulk_create() 不觸發 Signal
這是一個常見的陷阱——只有透過 Model 實例的 save() 和 delete() 方法才會觸發 Signal:
# 會觸發 post_save Signal
article = Article(title='Hello')
article.save()
# 不會觸發任何 Signal!
Article.objects.filter(category='tech').update(is_featured=True)
Article.objects.bulk_create([Article(title='A'), Article(title='B')])
Signal 測試策略
Signal 的隱式行為可能讓測試變得複雜。以下是三種常用的測試策略。
方式一:disconnect 後測試
from django.test import TestCase
from django.db.models.signals import post_save
from .signals import create_user_profile
from .models import User
class UserModelTest(TestCase):
def setUp(self):
# 斷開 Signal 避免副作用
post_save.disconnect(create_user_profile, sender=User)
def tearDown(self):
# 測試結束後重新連接
post_save.connect(create_user_profile, sender=User)
def test_user_creation_without_profile(self):
user = User.objects.create_user(username='test', password='pass')
self.assertEqual(user.username, 'test')
# UserProfile 不會被自動建立(已斷開 Signal)
方式二:使用 mock.patch
from unittest import mock
from django.test import TestCase
from .models import User
class SignalTest(TestCase):
@mock.patch('accounts.signals.create_user_profile')
def test_signal_is_called(self, mock_handler):
User.objects.create_user(username='test', password='pass')
# 驗證 Signal 接收者被呼叫了一次
mock_handler.assert_called_once()
方式三:factory_boy 的 mute_signals
import factory
from factory.django import mute_signals
from django.db.models.signals import post_save
@mute_signals(post_save)
def test_without_signals():
user = UserFactory()
# 此函式內的 post_save Signal 被靜音
總結
Django Signals 是一個強大的解耦工具,正確使用能讓你的應用程式架構更加模組化與靈活。以下整理本篇的核心要點:
- 觀察者模式 是 Signal 的設計基礎——發送者與接收者彼此不需要直接依賴,透過 Signal 作為中介傳遞事件通知
- 內建信號 涵蓋 Model 操作(pre_save、post_save、pre_delete、post_delete、m2m_changed)和 Request 生命週期(request_started、request_finished),滿足大部分使用場景
- @receiver 裝飾器 是連接 Signal 最推薦的方式,語法清晰且容易維護
- Signal.connect() 提供更高的彈性,支援動態連接與斷開,搭配
dispatch_uid可防止重複連接 - AppConfig.ready() 是載入 Signal 的正確位置,確保 App Registry 完全初始化後才進行連接
- 自訂 Signal 讓你可以定義應用程式專屬的事件通知,搭配
send_robust()避免單一接收者的錯誤影響其他接收者 - 避免反模式 ——不要在 Signal 中放業務邏輯、小心連鎖 Signal、耗時操作交給 Celery 非同步處理
Signal 是 Django 架構中的「幕後英雄」,用得好能讓程式碼更優雅,用不好則會讓除錯變成噩夢。關鍵原則是:Signal 只做事件通知與調度,業務邏輯交給 Service 層處理。