Django使用Signals監測model字段變化發送通知

上一篇文章《運維效率之數據遷移自動化》中講到了工單通知，本文將介紹工單通知實現過程當中的一些小技巧。全部演示均基於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 = ["dba1@ops-coffee.cn", "dba2@ops-coffee.cn"]

    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 = ["dba1@ops-coffee.cn", "dba2@ops-coffee.cn"]

    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：model初始化前觸發
   • post_init：model初始化後觸發
   • pre_save：save()方法前觸發
   • post_save：save()方法後觸發
   • pre_delete：delete()方法前觸發
   • post_delete：delete()方法後觸發
   • m2m_changed：ManyToManyField字段改變時觸發
   • class_prepared：沒用過字面意思理解吧
2. management signals
   • pre_migrate：migrate以前觸發
   • post_migrate：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'

綁定成功後就能夠在每次工單狀態發生變化時發送郵件了

---

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

• Django model轉字典的幾種方法
• Django配置Celery執行異步任務和定時任務