Skip to content

Django Signals#

Built-in Signals#

Model Signals#

# myapp/signals.py
from django.db.models.signals import pre_save, post_save, pre_delete, post_delete
from django.dispatch import receiver
from .models import Post

# Before saving
@receiver(pre_save, sender=Post)
def pre_save_post(sender, instance, **kwargs):
    # Auto-generate slug
    if not instance.slug:
        instance.slug = instance.title.lower().replace(' ', '-')

# After saving
@receiver(post_save, sender=Post)
def post_save_post(sender, instance, created, **kwargs):
    if created:
        print(f'New post created: {instance.title}')
        # Send email, create notification, etc.
    else:
        print(f'Post updated: {instance.title}')

# Before deleting
@receiver(pre_delete, sender=Post)
def pre_delete_post(sender, instance, **kwargs):
    # Delete associated files
    if instance.image:
        instance.image.delete(save=False)

# After deleting
@receiver(post_delete, sender=Post)
def post_delete_post(sender, instance, **kwargs):
    print(f'Post deleted: {instance.title}')

Connect Signals#

# myapp/apps.py
from django.apps import AppConfig

class MyappConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'

    def ready(self):
        import myapp.signals  # Import signals

Custom Signals#

# myapp/signals.py
from django.dispatch import Signal

# Define signal
post_published = Signal(providing_args=['post', 'user'])

# Send signal
from .signals import post_published

def publish_post(request, pk):
    post = get_object_or_404(Post, pk=pk)
    post.is_published = True
    post.save()

    # Send signal
    post_published.send(sender=Post, post=post, user=request.user)

    return redirect('post_detail', pk=pk)

# Receive signal
@receiver(post_published, sender=Post)
def handle_post_published(sender, post, user, **kwargs):
    # Send email notification
    send_email_notification(post, user)

    # Update cache
    cache.delete('published_posts')

    # Create activity log
    ActivityLog.objects.create(
        user=user,
        action='published',
        content_object=post
    )

Request/Response Signals#

# myapp/signals.py
from django.core.signals import request_started, request_finished
from django.dispatch import receiver

@receiver(request_started)
def request_started_handler(sender, **kwargs):
    # Log request start
    print(f'Request started: {kwargs.get("environ", {}).get("PATH_INFO")}')

@receiver(request_finished)
def request_finished_handler(sender, **kwargs):
    # Log request finish
    print('Request finished')

User Signals#

# myapp/signals.py
from django.contrib.auth.signals import user_logged_in, user_logged_out, user_login_failed
from django.dispatch import receiver

@receiver(user_logged_in)
def user_logged_in_handler(sender, request, user, **kwargs):
    # Update last login
    user.profile.last_login_ip = get_client_ip(request)
    user.profile.save()

    # Create login log
    LoginLog.objects.create(user=user, ip_address=get_client_ip(request))

@receiver(user_logged_out)
def user_logged_out_handler(sender, request, user, **kwargs):
    # Log logout
    print(f'User {user.username} logged out')

@receiver(user_login_failed)
def user_login_failed_handler(sender, credentials, **kwargs):
    # Log failed login attempt
    FailedLoginAttempt.objects.create(
        username=credentials.get('username'),
        ip_address=kwargs.get('request').META.get('REMOTE_ADDR')
    )

Multiple Receivers#

# Multiple functions can receive same signal
@receiver(post_save, sender=Post)
def send_email_on_post_save(sender, instance, created, **kwargs):
    if created:
        send_email(instance)

@receiver(post_save, sender=Post)
def update_cache_on_post_save(sender, instance, **kwargs):
    cache.delete('recent_posts')

@receiver(post_save, sender=Post)
def create_activity_on_post_save(sender, instance, created, **kwargs):
    Activity.objects.create(
        user=instance.author,
        action='created' if created else 'updated',
        post=instance
    )

Conditional Signals#

@receiver(post_save, sender=Post)
def handle_post_save(sender, instance, created, **kwargs):
    # Only for published posts
    if instance.is_published:
        # Send notification
        send_notification(instance)

    # Only on creation
    if created:
        # Welcome email
        send_welcome_email(instance.author)

Disconnect Signals#

from django.db.models.signals import post_save
from myapp.signals import handle_post_save

# Temporarily disconnect
post_save.disconnect(handle_post_save, sender=Post)

# Do something without signal

# Reconnect
post_save.connect(handle_post_save, sender=Post)

Best Practices#

  1. Import in ready() - Always import signals in AppConfig.ready()
  2. Use @receiver - Cleaner than manual connect()
  3. Check created - Use created parameter in post_save
  4. Avoid save() in signals - Can cause infinite loops
  5. Use update_fields - When saving in signals

Next: Celery & Celery Beat