文章

Django - 如何在 Django 中實現訊號(Signals)

Django 的 Signals(訊號) 是一種基於事件的工具,允許應用內部的組件互相通信,而不需要直接耦合。例如,當模型的某些操作(如保存或刪除)完成後,可以使用訊號自動觸發某些操作。


常見使用場景

  1. 記錄用戶操作:當用戶登入或登出時記錄日誌。
  2. 數據處理:在模型保存時自動更新某些欄位。
  3. 通知系統:在某些條件滿足時發送電子郵件或其他通知。

Django 中的訊號類型

Django 提供一些內建訊號,例如:

  1. 模型相關訊號
    • pre_save:模型保存之前觸發。
    • post_save:模型保存之後觸發。
    • pre_delete:模型刪除之前觸發。
    • post_delete:模型刪除之後觸發。
  2. 用戶相關訊號
    • user_logged_in:用戶成功登入時觸發。
    • user_logged_out:用戶登出時觸發。
    • user_login_failed:用戶登入失敗時觸發。
  3. 自定義訊號:開發者可創建自定義訊號。

如何使用訊號

1. 基本語法

  • 訊號使用 @receiver 裝飾器或直接調用 connect() 方法來連接。
  • 信號處理器是用來執行事件時的函數。

2. 示例:使用模型訊號

步驟 1:在模型保存時執行操作

tasks/models.py 中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver

class Task(models.Model):
    title = models.CharField(max_length=255)
    completed = models.BooleanField(default=False)

# 訊號處理器函數
@receiver(post_save, sender=Task)
def notify_task_saved(sender, instance, created, **kwargs):
    if created:
        print(f"New task created: {instance.title}")
    else:
        print(f"Task updated: {instance.title}")

步驟 2:測試訊號

新增或更新任務時,notify_task_saved 會被自動觸發:

1
python manage.py shell
1
2
3
4
5
6
7
8
from tasks.models import Task

# 創建新任務
task = Task.objects.create(title="Learn Django Signals")

# 更新任務
task.title = "Learn Django Signals - Updated"
task.save()

3. 自定義訊號

步驟 1:定義自定義訊號

tasks/signals.py 中:

1
2
3
4
from django.dispatch import Signal

# 自定義訊號
task_completed = Signal()

步驟 2:定義訊號處理器

tasks/signals.py 中:

1
2
3
4
5
6
from django.dispatch import receiver

@receiver(task_completed)
def handle_task_completed(sender, **kwargs):
    task = kwargs.get('task')
    print(f"Task completed: {task.title}")

步驟 3:觸發訊號

tasks/views.py 中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from django.shortcuts import get_object_or_404
from django.http import JsonResponse
from .models import Task
from .signals import task_completed

def complete_task(request, task_id):
    task = get_object_or_404(Task, id=task_id)
    task.completed = True
    task.save()

    # 觸發訊號
    task_completed.send(sender=Task, task=task)

    return JsonResponse({"message": f"Task '{task.title}' marked as completed."})

4. 在 Django 中正確管理訊號

步驟 1:集中管理訊號

可以在應用的 apps.py 中加載訊號:

1
2
3
4
5
6
7
8
from django.apps import AppConfig

class TasksConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'tasks'

    def ready(self):
        import tasks.signals  # 確保訊號在應用啟動時加載

訊號最佳實踐

  1. 避免邏輯耦合:訊號應該保持簡單,避免處理過多業務邏輯。
  2. 集中管理訊號:將訊號單獨存放於一個 signals.py 文件中。
  3. 避免循環引用:確保訊號加載順序正確,避免導致循環導入。
  4. 測試訊號功能:為訊號編寫單元測試,確保其正常觸發並執行預期邏輯。

訊號測試示例

tasks/tests.py 中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from django.test import TestCase
from django.dispatch import Signal
from django.dispatch import receiver

# 測試自定義訊號
class SignalTestCase(TestCase):
    def test_custom_signal(self):
        test_signal = Signal()

        # 定義接收器
        @receiver(test_signal)
        def test_receiver(sender, **kwargs):
            self.assertEqual(kwargs['message'], "Signal triggered!")

        # 觸發訊號
        test_signal.send(sender=self.__class__, message="Signal triggered!")

總結

  1. 訊號是事件驅動開發的強大工具,可用於監控模型操作、自定義事件通知等。
  2. Django 提供了許多內建訊號,適合處理常見的業務需求。
  3. 使用訊號時應遵循最佳實踐,確保代碼結構清晰且可維護。

作業

  1. 為用戶登入(user_logged_in)撰寫訊號,記錄用戶的登入時間。
  2. 為 Blog 系統添加訊號:在文章發佈時發送通知。
  3. 將訊號整合到 RESTful API 中,完成動態通知功能。

本文章以 CC BY 4.0 授權