Django使用Signals監測model欄位變化傳送通知

摘要： 上一篇文章 《運維效率之資料遷移自動化》 中講到了工單通知，本文將介紹工單通知實現過程中的一些小技巧。所有演示均基於Django2.0 閱讀此篇文章你可以： 解鎖一個python if的使用新姿勢 獲取一個利用signals做通知的真實案例 背景說明 先...

上一篇文章 ofollow,noindex" target="_blank">《運維效率之資料遷移自動化》 中講到了工單通知，本文將介紹工單通知實現過程中的一些小技巧。所有演示均基於Django2.0

閱讀此篇文章你可以：

• 解鎖一個python if的使用新姿勢
• 獲取一個利用signals做通知的真實案例

背景說明

先看看工單表簡化後的結構

class Ticket(models.Model):
'''工單表'''

STATE = (
(1, '待審批'),
(2, '已撤銷'),
(3, '已通過'),
(4, '被拒絕'),
(5, '已掛起'),
(6, '執行中'),
(7, '已完成'),
(8, '已失敗')
)

create_time = models.DateTimeField(auto_now_add=True, verbose_name='建立時間')
create_user = models.ForeignKey(User, on_delete=models.DO_NOTHING, verbose_name='建立使用者')

state = models.IntegerField(choices=STATE, default=1, verbose_name='工單狀態')

Ticket工單表有一個state欄位標識當前工單狀態，這個狀態會隨著工單的進行而改變，每當工單狀態改變時就需要傳送通知給相應的使用者，例如工單建立時，需要傳送給建立者一個工單建立成功的通知，同時傳送給稽核者一個待稽核的通知

通知

每一個狀態的變化都需要通知，為了程式碼易讀及解耦，我們需要寫一個單獨的通知類，當需要通知的時候呼叫一下就好了。通知類中需要判斷當前工單的狀態，那麼通常會寫成下邊這樣

class Notify:
def __init__(self):
self.dba_list = ["[email protected]", "[email protected]"]

def migration(self, pk):
'''遷移通知'''

_t = Ticket.objects.get(id=pk)
_u = _t.create_user.username
_s = _t.state

_d = "https://ops-coffee.cn/workflow/migration/%d/" % (_t.id)

if _s == 1:
try:
Email(
subject="[已提交]-[overmind]資料遷移工單",
content="你的資料遷移工單已提交，正在等待DBA審批，後續有狀態變更將會自動通知你。\r\n\r\n工單詳情：%s" % _d,
reciever_list=[_u]
)
except Exception as e:
print('Error:' + str(e))

try:
Email(
subject="[待審批]-[overmind]資料遷移工單",
content="你有工單需要審批，點選下方工單詳情連結及時審批。\r\n\r\n工單詳情：%s" % _d,
reciever_list=self.dba_list
)
except Exception as e:
print('Error:' + str(e))
elif _s == 6:
try:
Email(
subject="[執行中]-[overmind]資料遷移工單",
content="資料遷移工單已通過DBA稽核，正在執行中，後續有狀態變更將會自動通知你。\r\n\r\n工單詳情：%s" % _d,
reciever_list=[_u] + self.dba_list,
)
except Exception as e:
print('Error:' + str(e))
elif _s == 7:
try:
Email(
subject="[已完成]-[overmind]資料遷移工單",
content="資料遷移工單已自動完成遷移，請檢查最終狀態，如有任何疑問隨時聯絡DBA。\r\n\r\n工單詳情：%s" % _d,
reciever_list=[_u] + self.dba_list,
)
except Exception as e:
print('Error:' + str(e))

以上的程式碼可以看出來寫了很多重複的try程式碼，並且當狀態越多，if判斷越複雜時重複的程式碼也會越來越多，有沒有更簡潔的方式呢？

看看下邊的程式碼，維護一個狀態字典，然後用一個if判斷就可以實現上邊一堆if的程式碼寫法，是不是就簡潔了很多

class Notify:
def __init__(self):
self.dba_list = ["[email protected]", "[email protected]"]

def migration(self, pk):
'''遷移通知'''

_t = Ticket.objects.get(id=pk)
_u = _t.create_user.username
_s = _t.state

_d = "https://ops-coffee.cn/workflow/migration/%d/" %(_t.id)
smap = {
1: [{
"subject": "[已提交]-[overmind]資料遷移工單",
"content": "你的資料遷移工單已提交，正在等待DBA審批，後續有狀態變更將會自動通知你。\r\n\r\n工單詳情：%s" %_d,
"reciever_list": [_u],
}, {
"subject": "[待審批]-[overmind]資料遷移工單",
"content": "你有工單需要審批，點選下方工單詳情連結及時審批。\r\n\r\n工單詳情：%s" %_d,
"reciever_list": self.dba_list,
}],
6: [{
"subject": "[執行中]-[overmind]資料遷移工單",
"content": "資料遷移工單已通過DBA稽核，正在執行中，後續有狀態變更將會自動通知你。\r\n\r\n工單詳情：%s" %_d,
"reciever_list": [_u] + self.dba_list,
}],
7: [{
"subject": "[已完成]-[overmind]資料遷移工單",
"content": "資料遷移工單已自動完成遷移，請檢查最終狀態，如有任何疑問隨時聯絡DBA。\r\n\r\n工單詳情：%s" %_d,
"reciever_list": [_u] + self.dba_list,
}]
}

_list = smap[_s]
for i in range(0, len(_list)):
try:
Email(
subject=_list[i]['subject'], 
content=_list[i]['content'], 
reciever_list=_list[i]['reciever_list']
)
except Exception as e:
print('Error:' +str(e))

在構造字典的時候採用了狀態做key，通知變數做value，同時一個狀態可能會產生多個不同的通知，所以value採用列表的方式，這樣即可輕鬆實現一個狀態多條通知，每條通知都可以發給不同的人，有不同的主題，不同的內容。

Signals

上邊我們已經寫好了傳送通知的類，在view裡每次修改工單狀態之後呼叫下通知類即可實現通知傳送，但這樣通知跟view強耦合，且通知會分散在view中的多個地方，造成程式碼冗餘且不夠優雅。我們需要一個簡單優雅的方式來實現，signals可以說是非常有用了

Signals是Django自帶的一個訊號排程程式。如果你對svn或者git之類的hooks有了解，這個理解起來就簡單多了，通俗來說就是當你的程式產生一個事件時，會通過signals自動觸發其他的事件。就比如我們這個工單系統通知，當工單狀態產生變化時自動傳送郵件給相關人。

Django內部已經定義好了一些signal供我們使用，如果不能滿足我們也可以自定義signal，其中Django內部定義的signal主要分為幾類

1. model signals

   pre_init
   post_init
   pre_save
   post_save
   pre_delete
   post_delete
   m2m_changed
   class_prepared
   

2. management signals

   pre_migrate
   post_migrate
   

3. request/response signals

   request_started
   request_finished
   got_request_exception
   

4. test signals

   setting_changed
   template_rendered
   

5. Database Wrappers
   • connection_created ：連線建立時觸發

那麼訊號究竟該如何使用呢？下邊一個實際的例子來說明下訊號的使用

就以我們傳送通知的需求為例，workflow是一個普通的app，第一步需要新建 workflow/signals.py 檔案繫結signal

from django.db.models import signals
from django.dispatch import receiver

from workflow.models import Ticket
from workflow.backends.notify import Notify


@receiver(signals.post_init, sender=Ticket)
def migrate_notify_init(instance, **kwargs):
instance.old_state = instance.state


@receiver(signals.post_save, sender=Ticket)
def migrate_notify_post(instance, created, **kwargs):
if created or instance.old_state != instance.state:
Notify().migration(instance.id)

這裡用到了兩個signal， post_init 和 post_save

在model初始化之後通過 post_init 訊號獲取到state的值作為初始狀態值，在每次model執行save方法後呼叫 post_save 訊號獲取到新的狀態值，對兩次狀態值做比較如果不一致則表示狀態有更新發送通知

是上邊的判斷只能判斷到狀態變更了發通知，但工單在第一次建立時old_state和state是一樣的，所以也需要在save之後判斷下這次操作是不是新建，如果是新建同樣需要傳送通知

第二步載入signal，需要修改兩個配置檔案

config1： workflow/apps.py

from django.apps import AppConfig


class WorkflowConfig(AppConfig):
name = 'workflow'

def ready(self):
import workflow.signals

config2： workflow/__init__.py

default_app_config = 'workflow.apps.WorkflowConfig'

繫結成功後就可以在每次工單狀態發生變化時傳送郵件了

如果你覺得文章對你有幫助，請轉發分享給更多的人。如果你覺得讀的不盡興，推薦閱讀以下文章：

• CESHAB0dLgq7qZq5Q" rel="nofollow,noindex" target="_blank">Django model轉字典的幾種方法
• 5Gauysg" rel="nofollow,noindex" target="_blank">Django配置Celery執行非同步任務和定時任務