断开模型的信号并在django中重新连接

k3bvogb1  于 2023-05-19  发布在  Go
关注(0)|答案(6)|浏览(110)

我需要做一个保存与模型,但我需要断开一些接收器的信号之前保存它。
我是说
我有一个模型:

class MyModel(models.Model):
    ...

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

pre_save.connect(pre_save_model, sender=MyModel)

在代码的另一个地方,我需要这样的东西:

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

因为我需要在这种情况下保存模型而不执行函数pre_save_model。

sauutmhj

sauutmhj1#

对于一个干净和可重用的解决方案,您可以使用上下文管理器:

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
        )

现在,您可以执行以下操作:

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',
    )
  • 注意:如果你的信号处理器使用dispatch_uid,你必须使用dispatch_uid参数。*
lc8prwob

lc8prwob2#

你可以像Haystack在RealTimeSearchIndex中那样连接和断开信号,这似乎更标准:

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)
68de4m5k

68de4m5k3#

我还没有测试过下面的代码,但它应该可以工作:

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

它将使来自 * 所有 * 发送方的信号静音,但不仅仅是instance.__class__
此版本仅禁用给定模型的信号:

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
oalqel3c

oalqel3c4#

如果您只想断开和重新连接一个自定义信号,您可以使用以下代码:

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)

通过这种方式,你可以做到:

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

vktxenjb5#

我需要在单元测试中防止某些信号被触发,所以我基于qris的响应做了一个装饰器:

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

使用它很简单:

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

z31licg06#

这里是临时禁用每个示例的信号接收器的解决方案,允许在生产(线程安全)上使用它
[usage.py]**

from django.db.models.signals import post_save

payment = Payment()
with mute_signals_for(payment, signals=[post_save]):
   payment.save()  # handle_payment signal receiver will be skipped

[code.py]**

from contextlib import contextmanager
from functools import wraps

MUTE_SIGNALS_ATTR = '_mute_signals'

def mutable_signal_receiver(func):
    """Decorator for signals to allow to skip them by setting attr MUTE_SIGNALS_ATTR on instance,
    which can be done via mute_signals_for"""
    @wraps(func)
    def wrapper(sender, instance, signal, **kwargs):
        mute_signals = getattr(instance, MUTE_SIGNALS_ATTR, False)
        if mute_signals is True:
            pass  # skip all signals
        elif isinstance(mute_signals, list) and signal in mute_signals:
            pass  # skip user requested signal
        else:  # allow signal receiver
            return func(sender=sender, instance=instance, signal=signal, **kwargs)
    return wrapper

@contextmanager
def mute_signals_for(instance, signals):
    """Context manager to skip signals for @instance (django model), @signals can be
    True to skip all signals or list of specified signals, like [post_delete, post_save] """
    try:
        yield setattr(instance, MUTE_SIGNALS_ATTR, signals)
    finally:
        setattr(instance, MUTE_SIGNALS_ATTR, False)

[signals.py]**

@receiver(post_save, sender=Payment, dispatch_uid='post_payment_signal')
@mutable_signal_receiver
def handle_payment(sender, instance, created, **kwargs):
    """called after payment is registered in the system."""

相关问题