Django自定義訊號(signals)
阿新 • • 發佈:2018-12-15
django中自定義了一些singals,用於監聽一些操作,併發出通知
官方解釋:
Django 提供一個“訊號分發器”,允許解耦的應用在框架的其它地方發生操作時會被通知到。
簡單來說,訊號允許特定的sender通知一組receiver某些操作已經發生。這在多處程式碼和同一事件有關聯的情況下很有用。
django中已經內建了一些singals,在django/db/models/signal.py中,如
Model signals pre_init # django的modal執行其構造方法前,自動觸發 post_init # django的modal執行其構造方法後,自動觸發 pre_save # django的modal物件儲存前,自動觸發 post_save # django的modal物件儲存後,自動觸發 pre_delete # django的modal物件刪除前,自動觸發 post_delete # django的modal物件刪除後,自動觸發 m2m_changed # django的modal中使用m2m欄位操作第三張(add,remove,clear)前後,自動觸發 class_prepared # 程式啟動時,檢測已註冊的app中modal類,對於每一個類,自動觸發 Management signals pre_migrate # 執行migrate命令前,自動觸發 post_migrate # 執行migrate命令後,自動觸發 Request/response signals request_started # 請求到來前,自動觸發 request_finished # 請求結束後,自動觸發 got_request_exception # 請求異常後,自動觸發 Test signals setting_changed # 使用test測試修改配置檔案時,自動觸發 template_rendered # 使用test測試渲染模板時,自動觸發 Database Wrappers connection_created # 建立資料庫連線時,自動觸發
用法:
利用這幾個singals可以實現model中的一些聯動操作,比如,要想更改通過model更新記錄時,記下操作者的日誌,可以直接在操作的地方使用post_save裝飾器,
或者改寫post_save,使其記錄相關資訊,一勞永逸。或者在request請求時,記錄請求資訊。
from django.core.signals import request_finished from django.dispatch import receiver @receiver(request_finished) def my_callback(sender, **kwargs): print("Request finished!")
如何自定義singals?
a. 定義singal檔案
import django.dispatch
pizza_done = django.dispatch.Signal(providing_args=["toppings", "size"])
b. 註冊singal
def callback(sender, **kwargs):
print("callback")
print(sender,kwargs)
pizza_done.connect(callback)
c. 觸發訊號
from 路徑 import pizza_done pizza_done.send(sender='seven',toppings=123, size=456)
需求場景:
專案中有一個需求,當model(即庫的資料)被修改或者刪除時,自動觸發一個redis的同步任務(後來發現這個需求沒有意義…),model的儲存有post_save,刪除有post_delete,唯獨沒有update,而程式碼中使用update的場景蠻多的,就搜了下為什麼就是沒有update的singals。
其實很早就有人給django官方提過這種方式,為什麼不在官方版本中新增,具體這個pr為什麼沒有被接受,可以看下里面的討論,反正當時的django1.9仍然不支援,只能自己先寫一個用用,有問題了再撤掉好了。
解決方式:
singals.py檔案
# coding:utf-8
from django.dispatch import Signal
post_update = Signal(providing_args=["user"])
models.py檔案
-----------針對某個model,重寫其queryset中的update方法-----------
//引入自定義的signal檔案
from tools import signals
class MyCustomQuerySet(models.query.QuerySet):
def update(self, **kwargs):
super(MyCustomQuerySet, self).update(**kwargs)
//update被呼叫時, 傳送該singalsignals
signals.post_update.send(sender=self.model, user="xxx")
print("finished!")
class MyCustomManager(models.Manager):
def get_queryset(self):
return MyCustomQuerySet(self.model, using=self._db)
class crontab_ping(models.Model):
name = models.CharField(max_length=64, blank=True, null=True)
objects = MyCustomManager()
callback.py檔案:
-------接收signal,觸發操作----------
from tools.signals import post_update
@receiver(post_update)
def post_update_callback(sender, **kwargs):
print(kwargs['user'])
print("post_update_success")