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执行异步任务和定时任务