Django Signals

# Django Signals Implementation

# 1. Built-in Signals
from django.db.models.signals import pre_save, post_save, pre_delete, post_delete
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import Post, Comment, Like

# Post save signal
@receiver(post_save, sender=Post)
def post_save_handler(sender, instance, created, **kwargs):
    """Handle post save events"""
    if created:
        # New post created
        print(f"New post created: {instance.title}")
        
        # Update user post count
        instance.author.profile.post_count += 1
        instance.author.profile.save()
        
        # Send notification
        send_post_notification(instance)
    else:
        # Post updated
        print(f"Post updated: {instance.title}")
        
        # Update search index
        update_search_index(instance)

# Post delete signal
@receiver(post_delete, sender=Post)
def post_delete_handler(sender, instance, **kwargs):
    """Handle post delete events"""
    # Update user post count
    instance.author.profile.post_count -= 1
    instance.author.profile.save()
    
    # Remove from search index
    remove_from_search_index(instance)
    
    # Clean up related files
    cleanup_post_files(instance)

# Comment signal
@receiver(post_save, sender=Comment)
def comment_save_handler(sender, instance, created, **kwargs):
    """Handle comment save events"""
    if created:
        # Update post comment count
        instance.post.comment_count += 1
        instance.post.save()
        
        # Send notification to post author
        send_comment_notification(instance)
        
        # Update post activity timestamp
        instance.post.last_activity = timezone.now()
        instance.post.save()

# 2. Custom Signals
from django.dispatch import Signal

# Define custom signals
post_published = Signal()
post_viewed = Signal()
user_registered = Signal()
post_liked = Signal()

# Signal handlers
@receiver(post_published)
def handle_post_published(sender, post, **kwargs):
    """Handle post publication"""
    # Send email notifications
    send_publication_notifications(post)
    
    # Update RSS feed
    update_rss_feed()
    
    # Post to social media
    post_to_social_media(post)
    
    # Update search index
    update_search_index(post)

@receiver(post_viewed)
def handle_post_viewed(sender, post, user, **kwargs):
    """Handle post view"""
    # Increment view count
    post.view_count += 1
    post.save()
    
    # Track user activity
    track_user_activity(user, 'post_view', post)
    
    # Update trending posts
    update_trending_posts(post)

@receiver(user_registered)
def handle_user_registered(sender, user, **kwargs):
    """Handle new user registration"""
    # Send welcome email
    send_welcome_email(user)
    
    # Create user profile
    create_user_profile(user)
    
    # Add to mailing list
    add_to_mailing_list(user)
    
    # Send onboarding sequence
    start_onboarding_sequence(user)

@receiver(post_liked)
def handle_post_liked(sender, post, user, **kwargs):
    """Handle post like"""
    # Update like count
    post.like_count += 1
    post.save()
    
    # Send notification to post author
    send_like_notification(post, user)
    
    # Update user activity
    track_user_activity(user, 'post_like', post)

# 3. Advanced Signal Patterns
class SignalManager:
    """Advanced signal management with error handling and logging"""
    
    def __init__(self):
        self.signal_handlers = {}
        self.error_handlers = {}
    
    def register_handler(self, signal, handler, error_handler=None):
        """Register a signal handler with optional error handling"""
        if signal not in self.signal_handlers:
            self.signal_handlers[signal] = []
        
        self.signal_handlers[signal].append(handler)
        
        if error_handler:
            self.error_handlers[handler] = error_handler
    
    def dispatch_signal(self, signal, **kwargs):
        """Dispatch a signal with error handling"""
        if signal not in self.signal_handlers:
            return
        
        for handler in self.signal_handlers[signal]:
            try:
                handler(**kwargs)
            except Exception as e:
                # Log error
                logger.error(f"Signal handler {handler.__name__} failed: {e}")
                
                # Call error handler if available
                if handler in self.error_handlers:
                    try:
                        self.error_handlers[handler](e, **kwargs)
                    except Exception as error_e:
                        logger.error(f"Error handler also failed: {error_e}")
    
    def unregister_handler(self, signal, handler):
        """Unregister a signal handler"""
        if signal in self.signal_handlers and handler in self.signal_handlers[signal]:
            self.signal_handlers[signal].remove(handler)

# Usage
signal_manager = SignalManager()

# Register handlers
signal_manager.register_handler(
    post_published,
    handle_post_published,
    error_handler=lambda e, **kwargs: logger.error(f"Publication failed: {e}")
)

# Dispatch signals
signal_manager.dispatch_signal(post_published, post=post_instance)

# 4. Signal Decorators and Utilities
from functools import wraps
from django.core.cache import cache

def signal_cache_invalidator(cache_pattern):
    """Decorator to invalidate cache when signals are triggered"""
    def decorator(func):
        @wraps(func)
        def wrapper(sender, instance, **kwargs):
            # Execute the original function
            result = func(sender, instance, **kwargs)
            
            # Invalidate cache
            cache_keys = cache_pattern(instance)
            for key in cache_keys:
                cache.delete(key)
            
            return result
        return wrapper
    return decorator

def get_cache_keys_for_post(post):
    """Generate cache keys for a post"""
    return [
        f'post_{post.id}',
        f'user_posts_{post.author.id}',
        f'category_posts_{post.category.id}',
        'recent_posts',
        'trending_posts'
    ]

# Use the decorator
@signal_cache_invalidator(get_cache_keys_for_post)
@receiver(post_save, sender=Post)
def invalidate_post_cache(sender, instance, **kwargs):
    """Invalidate post-related cache"""
    pass

# 5. Signal Testing
from django.test import TestCase
from unittest.mock import patch

class SignalTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser',
            password='testpass123'
        )
        self.post = Post.objects.create(
            title='Test Post',
            content='Test content',
            author=self.user
        )
    
    @patch('myapp.signals.send_publication_notifications')
    def test_post_published_signal(self, mock_send_notifications):
        """Test that post published signal is sent"""
        # Publish the post
        self.post.status = 'published'
        self.post.save()
        
        # Check if signal was sent
        mock_send_notifications.assert_called_once_with(self.post)
    
    def test_post_delete_signal(self):
        """Test that post delete signal is sent"""
        initial_count = self.user.profile.post_count
        
        # Delete the post
        self.post.delete()
        
        # Check if post count was updated
        self.user.profile.refresh_from_db()
        self.assertEqual(self.user.profile.post_count, initial_count - 1)

# 6. Performance Considerations
class OptimizedSignalHandler:
    """Signal handler with performance optimizations"""
    
    def __init__(self):
        self.batch_size = 100
        self.pending_updates = []
    
    @receiver(post_save, sender=Post)
    def batch_update_search_index(self, sender, instance, **kwargs):
        """Batch update search index for better performance"""
        self.pending_updates.append(instance.id)
        
        if len(self.pending_updates) >= self.batch_size:
            self._process_batch()
    
    def _process_batch(self):
        """Process batch of updates"""
        if not self.pending_updates:
            return
        
        # Process in batch
        post_ids = self.pending_updates[:self.batch_size]
        self.pending_updates = self.pending_updates[self.batch_size:]
        
        # Update search index in batch
        update_search_index_batch(post_ids)

# Usage
optimized_handler = OptimizedSignalHandler()