Chapter 12: Middleware and Signals

12.2 Django Signals

Signal System Concept

The Django signal system allows certain senders to notify a set of receivers that some operation has occurred. Signals can notify decoupled applications when certain code is executed. Signals are a way to implement the observer pattern in Django.

Main uses of signals:

1. Execute certain operations when models are saved or deleted
2. Execute certain operations when users log in or log out
3. Execute certain operations during request processing
4. Implement decoupled communication between applications

Built-in Signals

Django provides many built-in signals:

# Common built-in signals
from django.db.models.signals import pre_save, post_save, pre_delete, post_delete
from django.contrib.auth.signals import user_logged_in, user_logged_out, user_login_failed
from django.core.signals import request_started, request_finished
from django.db.backends.signals import connection_created

# Model save signals
@receiver(pre_save, sender=MyModel)
def my_model_pre_save(sender, instance, **kwargs):
    """Execute before model is saved"""
    print(f"About to save {instance}")

@receiver(post_save, sender=MyModel)
def my_model_post_save(sender, instance, created, **kwargs):
    """Execute after model is saved"""
    if created:
        print(f"Newly created {instance}")
    else:
        print(f"Updated {instance}")

# Model delete signals
@receiver(pre_delete, sender=MyModel)
def my_model_pre_delete(sender, instance, **kwargs):
    """Execute before model is deleted"""
    print(f"About to delete {instance}")

@receiver(post_delete, sender=MyModel)
def my_model_post_delete(sender, instance, **kwargs):
    """Execute after model is deleted"""
    print(f"Deleted {instance}")

# User authentication signals
@receiver(user_logged_in)
def user_logged_in_callback(sender, request, user, **kwargs):
    """Execute when user logs in"""
    print(f"{user.username} logged in")
    # Record login log
    LoginLog.objects.create(user=user, action='login')

@receiver(user_logged_out)
def user_logged_out_callback(sender, request, user, **kwargs):
    """Execute when user logs out"""
    print(f"{user.username} logged out")
    # Record logout log
    LoginLog.objects.create(user=user, action='logout')

@receiver(user_login_failed)
def user_login_failed_callback(sender, credentials, request, **kwargs):
    """Execute when user login fails"""
    print(f"Login failed: {credentials.get('username')}")
    # Record failure log
    FailedLoginLog.objects.create(
        username=credentials.get('username'),
        ip_address=request.META.get('REMOTE_ADDR')
    )

# Request processing signals
@receiver(request_started)
def request_started_callback(sender, **kwargs):
    """Execute when request starts"""
    print("Request started processing")

@receiver(request_finished)
def request_finished_callback(sender, **kwargs):
    """Execute when request completes"""
    print("Request processing completed")

Custom Signals

Creating and using custom signals:

# signals.py
import django.dispatch

# Define custom signals
payment_completed = django.dispatch.Signal(providing_args=["order", "amount", "user"])
user_profile_updated = django.dispatch.Signal(providing_args=["user", "old_profile", "new_profile"])
email_sent = django.dispatch.Signal(providing_args=["recipient", "subject", "template"])

# In Django 3.1+, it's recommended not to specify providing_args
payment_completed = django.dispatch.Signal()
user_profile_updated = django.dispatch.Signal()
email_sent = django.dispatch.Signal()

# Sending signals
# Method 1: Using send()
payment_completed.send(sender=None, order=order, amount=amount, user=user)

# Method 2: Using send_robust() (safer)
payment_completed.send_robust(sender=None, order=order, amount=amount, user=user)

# models.py
from django.db import models
from django.contrib.auth.models import User
from .signals import payment_completed

class Order(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20, default='pending')
    created_at = models.DateTimeField(auto_now_add=True)
    
    def complete_payment(self):
        """Complete payment"""
        self.status = 'completed'
        self.save()
        
        # Send payment completion signal
        payment_completed.send(
            sender=self.__class__,
            order=self,
            amount=self.amount,
            user=self.user
        )

# receivers.py
from django.dispatch import receiver
from .signals import payment_completed, user_profile_updated
from .models import Order, UserProfile

@receiver(payment_completed)
def handle_payment_completed(sender, order, amount, user, **kwargs):
    """Handle payment completion"""
    # Send confirmation email
    send_payment_confirmation_email(user, order)
    
    # Update user points
    user.profile.points += int(amount)
    user.profile.save()
    
    # Record log
    PaymentLog.objects.create(
        user=user,
        order=order,
        amount=amount,
        status='completed'
    )

@receiver(user_profile_updated)
def handle_profile_updated(sender, user, old_profile, new_profile, **kwargs):
    """Handle user profile update"""
    # Check for important information changes
    if old_profile.email != new_profile.email:
        # Send email verification
        send_email_verification(user, new_profile.email)
    
    # Record change log
    ProfileChangeLog.objects.create(
        user=user,
        old_data=old_profile.__dict__,
        new_data=new_profile.__dict__
    )

Signal Handlers

Implementation and best practices of signal handlers:

# apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'
    
    def ready(self):
        """Register signal handlers when app is ready"""
        import myapp.signals  # Ensure signal handlers are loaded

# signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from django.contrib.auth.models import User
from .models import Article, Comment

# Use weak=False to ensure signal handlers are not garbage collected
@receiver(post_save, sender=Article, weak=False)
def invalidate_article_cache(sender, instance, **kwargs):
    """Clear cache when article is saved"""
    cache.delete(f'article_{instance.pk}')
    cache.delete('article_list')
    print(f"Cleared cache for article {instance.pk}")

# Conditional signal reception
@receiver(post_save, sender=Comment)
def notify_on_comment(sender, instance, created, **kwargs):
    """Send notification on new comment"""
    if created and instance.article.author != instance.author:
        # Only send notification for new comments and when commenter is not the article author
        send_notification(
            recipient=instance.article.author,
            message=f"Your article '{instance.article.title}' has a new comment"
        )

# Use sender parameter to receive signals from multiple models
@receiver(post_save, sender=Article)
@receiver(post_save, sender=Comment)
def update_last_modified(sender, instance, **kwargs):
    """Update last modified time"""
    from django.utils import timezone
    # Update site last modified time
    SiteSettings.objects.update(last_modified=timezone.now())

# Asynchronous signal handlers
import threading

@receiver(post_save, sender=Article)
def send_email_async(sender, instance, created, **kwargs):
    """Send email asynchronously"""
    if created:
        # Send email in new thread to avoid blocking main request
        email_thread = threading.Thread(
            target=send_welcome_email,
            args=(instance.author.email,)
        )
        email_thread.daemon = True
        email_thread.start()

# Exception handling in signal handlers
@receiver(post_save, sender=Article)
def risky_signal_handler(sender, instance, **kwargs):
    """Signal handler that may fail"""
    try:
        # Operations that may fail
        external_api_call(instance)
    except Exception as e:
        # Log error but don't interrupt other signal handlers
        import logging
        logger = logging.getLogger(__name__)
        logger.error(f"Signal handler error: {e}")
        # Can choose to re-raise exception or handle silently
        # raise  # Re-raising will interrupt other signal handlers

# Use dispatch_uid to avoid duplicate registration
@receiver(post_save, sender=Article, dispatch_uid="unique_article_handler")
def unique_article_handler(sender, instance, **kwargs):
    """Signal handler that ensures only one registration"""
    print("Processing article save")

# Conditional signal handlers
from django.conf import settings

@receiver(post_save, sender=Article)
def production_only_handler(sender, instance, **kwargs):
    """Execute only in production environment"""
    if not settings.DEBUG:
        # Production environment specific processing
        send_to_analytics(instance)

# Database transactions in signal handlers
from django.db import transaction

@receiver(post_save, sender=Article)
def transactional_handler(sender, instance, **kwargs):
    """Transactional signal handler"""
    with transaction.atomic():
        try:
            # Execute related operations
            update_statistics(instance)
            create_audit_log(instance)
        except Exception as e:
            # Rollback transaction and log error
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Transactional signal handler error: {e}")
            raise  # Re-raise exception

# Best practice examples for signal handlers
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    """Create user profile"""
    if created:
        # Create profile when new user is created
        from .models import UserProfile
        UserProfile.objects.create(user=instance)

@receiver(post_delete, sender=Article)
def cleanup_article_files(sender, instance, **kwargs):
    """Clean up related files when article is deleted"""
    # Delete article-related uploaded files
    if instance.image:
        instance.image.delete(save=False)
    
    # Clean up cache
    cache.delete_many([
        f'article_{instance.pk}',
        'article_list',
        f'user_articles_{instance.author.pk}'
    ])

# Complex signal handler example
@receiver(post_save, sender=Article)
def complex_article_handler(sender, instance, created, **kwargs):
    """Complex article processing logic"""
    # 1. Update search index
    update_search_index.delay(instance.pk)  # Use Celery async task
    
    # 2. Send notification
    if created:
        send_notification_to_followers(instance.author, instance)
    
    # 3. Update statistics
    update_user_stats(instance.author)
    
    # 4. Generate thumbnail (if image article)
    if instance.image and created:
        generate_thumbnail.delay(instance.image.path)
    
    # 5. Record activity log
    ActivityLog.objects.create(
        user=instance.author,
        action='created' if created else 'updated',
        content_type='article',
        object_id=instance.pk
    )

By using the signal system appropriately, you can achieve decoupling between applications, making code more modular and maintainable.

Summary

The Django signal system provides a powerful decoupling mechanism:

1. ✅ Understand the basic concepts and working principles of the signal system
2. ✅ Master the usage of common built-in signals
3. ✅ Be able to create and use custom signals
4. ✅ Implement signal handlers to process specific business logic
5. ✅ Follow best practices for signal handling

Proper use of the signal system can build more flexible and maintainable Django applications.

Next Article

We will learn about Django project实战 development.

13.1 Project Planning →

Directory

Return to Course Directory