Disconnect signals for models and reconnect in django

Question

I need make a save with a model but i need disconnect some receivers of the signals before save it.

I mean,

I have a model:

class MyModel(models.Model):
    ...

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

pre_save.connect(pre_save_model, sender=MyModel)

and in another place in the code i need something like:

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

Because i need in this case, save the model without execute the function pre_save_model.

Dumbstruck

Answer 1

For a clean and reusable solution, you can use a context manager:

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
        )

Now, you can do something like the following:

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',
    )

Note: If your signal handler uses a dispatch_uid, you MUST use the dispatch_uid arg.

Gerek

Great. This is the most elegant solution. You can reuse the context manager in several parts of the code. – Lenni

A small warning: weak=False is not the default when connecting a receiver to a signal. – Trantrance

weak is deprecated Also, people should be aware that disabling a signal will prevent all instances from triggering the signal, not just current context (i.e. other threads, as signals seem to be thread safe), as suggested here – Epispastic

@DanielDubovski They seem to be thread-safe? Then it should work like this, won't it? – Narcosynthesis

If you have multiple receiver decorators on the signal, it doesn't work for some reason. But still very nice solution! – Narcosynthesis

Be careful with this code as it is not thread safe. We modify the global state of the application @Gerek #20908131 – Knapp

weak is only deprecated on the disconnect method – Occlusion

Answer 2

You can connect and disconnect signals as Haystack does in RealTimeSearchIndex, which seems more standard:

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)

Interpretative

pre_savel_model is the same as pre_save? – Abri

@Abri - I'm assuming that pre_save_model is just an example of a signal receiver name. The first argument to connect or disconnect is the signal receiver. (Docs) – Verbosity

Be careful with this code as it is not thread safe. We modify the global state of the application @Interpretative #20908131 – Knapp

Answer 3

I haven't tested the following code, but it should work:

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

It will silence signals from all sender's though not just instance.__class__.

---

This version disables only the given model's signals:

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

Africa

You should probably wrap the save in a try block and the re-attachment of receivers in a finally bloc. Otherwise you might disconnect the signals forever. – Anodize

yeah use try..finally – Pelasgian

Be careful with this code as it is not thread safe. We modify the global state of the application @Africa #20908131 – Knapp

Answer 4

If you only want disconnect and reconnect one custom signal, you may use this code:

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)

In this way you can make this:

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

Dumbstruck

Be careful with this code as it is not thread safe. We modify the global state of the application @Dumbstruck #20908131 – Knapp

Answer 5

I needed to prevent certain signals from firing during unittests so I made a decorator based on qris's response:

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

Using it is simple:

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

Analeptic

Disabling signals during tests kinda misses the point of testing. Code flow should stay the same regarding of scenario. If there is code you don't need to execute as part of the test, then mock it's result, don't skip it. – Epispastic

If the wrapped function is meant to return some value, your code won't work. You must return the function result value in your decorator. – Teamster

@DanielDubovski there are cases where you might have a section of test code that is generating a lot of test data. Normally, if a user created these models it would have a side effect, but you want to skip that for now. Yes, you could mock all of the receiver functions, but at that point it will be more explicit if you just disable the signals. Then you would create a normal integration test where the signals are re-enabled. – Chili

@JordanReiter I understand your point, but, I still disagree. IMHO, changing code flow for testing purposes is a bad practice, as it is easy to forget about the different paths the code can take. specifically, mocking is very explicit in nature and in my opinion more pythonic. That said, like with any rule, I guess there will always be an exception to a rule of thumb, the danger is that the exception will become the norm for later maintainers.. – Epispastic

Answer 6

Here is solution to temporary disable signal receiver per instance which allows to use it on production (thread-safe)

[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."""

Merits