I have a signal callback in django:
@receiver(post_save, sender=MediumCategory)
def update_category_descendants(sender, **kwargs):
def children_for(category):
return MediumCategory.objects.filter(parent=category)
def do_update_descendants(category):
children = children_for(category)
descendants = list() + list(children)
for descendants_part in [do_update_descendants(child) for child in children]:
descendants += descendants_part
category.descendants.clear()
for descendant in descendants:
if category and not (descendant in category.descendants.all()):
category.descendants.add(descendant)
category.save()
return list(descendants)
# call it for update
do_update_descendants(None)
...but in the signal handler's body I'm using .save()
on the same model MediumCategory
. This causes the signal to be dispatched again. How can I disable it?
The perfect solution would be a with
statement with some 'magic' inside.
UPDATE: Here is my final solution, if anyone interested:
class MediumCategory(models.Model):
name = models.CharField(max_length=100)
slug = models.SlugField(blank=True)
parent = models.ForeignKey('self', blank=True, null=True)
parameters = models.ManyToManyField(AdvertisementDescriptonParameter, blank=True)
count_mediums = models.PositiveIntegerField(default=0)
count_ads = models.PositiveIntegerField(default=0)
descendants = models.ManyToManyField('self', blank=True, null=True)
def save(self, *args, **kwargs):
self.slug = slugify(self.name)
super(MediumCategory, self).save(*args, **kwargs)
def __unicode__(self):
return unicode(self.name)
(...)
@receiver(post_save, sender=MediumCategory)
def update_category_descendants(sender=None, **kwargs):
def children_for(category):
return MediumCategory.objects.filter(parent=category)
def do_update_descendants(category):
children = children_for(category)
descendants = list() + list(children)
for descendants_part in [do_update_descendants(child) for child in children]:
descendants += descendants_part
if category:
category.descendants.clear()
for descendant in descendants:
category.descendants.add(descendant)
return list(descendants)
# call it for update
do_update_descendants(None)
To disable a signal on your model, a simple way to go is to set an attribute on the current instance to prevent upcoming signals firing.
This can be done using a simple decorator that checks if the given instance has the 'skip_signal' attribute, and if so prevents the method from being called:
from functools import wraps
def skip_signal():
def _skip_signal(signal_func):
@wraps(signal_func)
def _decorator(sender, instance, **kwargs):
if hasattr(instance, 'skip_signal'):
return None
return signal_func(sender, instance, **kwargs)
return _decorator
return _skip_signal
You can now use it this way:
from django.db.models.signals import post_save
from django.dispatch import receiver
@receiver(post_save, sender=MyModel)
@skip_signal()
def my_model_post_save(sender, instance, **kwargs):
instance.some_field = my_value
# Here we flag the instance with 'skip_signal'
# and my_model_post_save won't be called again
# thanks to our decorator, avoiding any signal recursion
instance.skip_signal = True
instance.save()
Hope This helps.
Perhaps I'm wrong, but I think that category.save()
is not needed in your code, add() is enough because change is made in descendant but in category.
Also, to avoid signals you can:
Descendant.objects.filter( pk = descendant.pk ).update( category = category )
Here is solution to temporary disable signal receiver per instance which allows to use it on production (bc it is 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."""
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With