Ngắt kết nối tín hiệu cho mô hình và kết nối lại trong django

Question

Tôi cần thực hiện lưu với mô hình nhưng tôi cần ngắt kết nối một số bộ thu tín hiệu trước khi lưu.

Ý tôi là,

Tôi có một mô hình:

class MyModel(models.Model): 
    ... 

def pre_save_model(sender, instance, **kwargs): 
    ... 

pre_save.connect(pre_save_model, sender=MyModel) 

và ở một nơi khác trong mã tôi cần một cái gì đó như:

a = MyModel() 
... 
disconnect_signals_for_model(a) 
a.save() 
... 
reconnect_signals_for_model(a) 

Bởi vì tôi cần trong trường hợp này, tiết kiệm mô hình mà không thực thi hàm pre_save_model.

Answer 1

Đối với một giải pháp làm sạch và tái sử dụng, bạn có thể sử dụng một trình quản lý nội dung:

class temp_disconnect_signal(): 
    """ Temporarily disconnect a model from a signal """ 
    def __init__(self, signal, receiver, sender, dispatch_uid=None): 
     self.signal = signal 
     self.receiver = receiver 
     self.sender = sender 
     self.dispatch_uid = dispatch_uid 

    def __enter__(self): 
     self.signal.disconnect(
      receiver=self.receiver, 
      sender=self.sender, 
      dispatch_uid=self.dispatch_uid, 
      weak=False 
     ) 

    def __exit__(self, type, value, traceback): 
     self.signal.connect(
      receiver=self.receiver, 
      sender=self.sender, 
      dispatch_uid=self.dispatch_uid, 
      weak=False 
     ) 

Bây giờ, bạn có thể làm điều gì đó như sau:

from django.db.models import signals 

from your_app.signals import some_receiver_func 
from your_app.models import SomeModel 

... 
kwargs = { 
    'signal': signals.post_save, 
    'receiver': some_receiver_func, 
    'sender': SomeModel, 
    'dispatch_uid': "optional_uid" 
} 
with temp_disconnect_signal(**kwargs): 
    SomeModel.objects.create(
     name='Woohoo', 
     slug='look_mom_no_signals', 
    ) 

Lưu ý: Nếu xử lý tín hiệu của bạn sử dụng một dispatch_uid, bạn PHẢI sử dụng dispatch_uid arg.

Answer 2

tôi đã không kiểm tra đoạn mã sau, nhưng nó cũng làm việc:

from django.db.models.signals import pre_save 


def save_without_the_signals(instance, *args, **kwargs): 
    receivers = pre_save.receivers 
    pre_save.receivers = [] 
    new_instance = instance.save(*args, **kwargs) 
    pre_save.receivers = receivers 
    return new_instance 

Nó sẽ im lặng tín hiệu từ tất cả người gửi mặc dù không chỉ instance.__class__.

---

Phiên bản này sẽ vô hiệu hóa các tín hiệu chỉ có mô hình cho trước:

from django.db.models.signals import pre_save 
from django.dispatch.dispatcher import _make_id 


def save_without_the_signals(instance, *args, **kwargs): 
    receivers = [] 
    sender_id = _make_id(instance.__class__) 
    for index in xrange(len(self.receivers)): 
     if pre_save.receivers[index][0][1] == sender_id: 
      receivers.append(pre_save.receivers.pop(index)) 
    new_instance = instance.save(*args, **kwargs) 
    pre_save.receivers.extend(receivers) 
    return new_instance 

Answer 3

Nếu bạn chỉ muốn ngắt kết nối và kết nối lại một tín hiệu tùy chỉnh, bạn có thể sử dụng mã này:

def disconnect_signal(signal, receiver, sender): 
    disconnect = getattr(signal, 'disconnect') 
    disconnect(receiver, sender) 

def reconnect_signal(signal, receiver, sender): 
    connect = getattr(signal, 'connect') 
    connect(receiver, sender=sender) 

Bằng cách này bạn có thể thực hiện việc này:

disconnect_signal(pre_save, pre_save_model, MyModel) 
a.save() 
reconnect_signal(pre_save, pre_save_model, MyModel) 

Answer 4

Bạn có thể kết nối và tín hiệu ngắt kết nối như Haystack làm trong RealTimeSearchIndex, mà dường như nhiều tiêu chuẩn:

from django.db.models import signals 
signals.pre_save.disconnect(pre_save_model, sender=MyModel) 
a.save() 
signals.pre_save.connect(pre_save_model, sender=MyModel) 

Answer 5

tôi cần thiết để ngăn chặn các tín hiệu nhất định từ bắn trong unittests vì vậy tôi đã trang trí dựa trên phản ứng QRIS của:

from django.db.models import signals 

def prevent_signal(signal_name, signal_fn, sender): 
    def wrap(fn): 
     def wrapped_fn(*args, **kwargs): 
      signal = getattr(signals, signal_name) 
      signal.disconnect(signal_fn, sender) 
      fn(*args, **kwargs) 
      signal.connect(signal_fn, sender) 
     return wrapped_fn 
    return wrap 

Sử dụng nó rất đơn giản:

@prevent_signal('post_save', my_signal, SenderClass) 
def test_something_without_signal(self): 
    # the signal will not fire inside this test