Django signals provide a decoupled way to allow certain senders to notify a set of receivers when some actions have taken place. They're particularly useful for performing actions when models are saved, deleted, or when other events occur in your Django application.
Signals are a form of the observer pattern implementation in Django. They allow certain senders to notify receivers when specific actions occur. This enables loose coupling between different parts of your application.
# Model signals
from django.db.models.signals import (
pre_init, post_init,
pre_save, post_save,
pre_delete, post_delete,
m2m_changed
)
# Request/response signals
from django.core.signals import (
request_started, request_finished,
got_request_exception
)
# Management signals
from django.db.models.signals import (
pre_migrate, post_migrate
)
# Test signals
from django.test.signals import (
setting_changed, template_rendered
)
# models.py
from django.db import models
from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from django.contrib.auth.models import User
class UserProfile(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
bio = models.TextField(blank=True)
avatar = models.ImageField(upload_to='avatars/', blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
# Method 1: Using @receiver decorator
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
"""Create a UserProfile when a User is created"""
if created:
UserProfile.objects.create(user=instance)
@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
"""Save the UserProfile when User is saved"""
if hasattr(instance, 'userprofile'):
instance.userprofile.save()
# Method 2: Using connect() method
def user_logged_in_handler(sender, request, user, **kwargs):
"""Handle user login"""
print(f"User {user.username} logged in from {request.META.get('REMOTE_ADDR')}")
from django.contrib.auth.signals import user_logged_in
user_logged_in.connect(user_logged_in_handler)
# signals.py - Dedicated signals module
from django.db.models.signals import post_save, pre_delete, m2m_changed
from django.dispatch import receiver
from django.core.cache import cache
from django.core.mail import send_mail
from django.conf import settings
import logging
logger = logging.getLogger(__name__)
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey(User, on_delete=models.CASCADE)
tags = models.ManyToManyField('Tag')
published = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Tag(models.Model):
name = models.CharField(max_length=50, unique=True)
slug = models.SlugField(unique=True)
@receiver(post_save, sender=Post)
def post_saved_handler(sender, instance, created, **kwargs):
"""Handle post save events"""
# Log the action
action = "created" if created else "updated"
logger.info(f"Post '{instance.title}' was {action} by {instance.author.username}")
# Clear cache
cache_key = f"post_{instance.id}"
cache.delete(cache_key)
cache.delete("recent_posts")
# Send notification email for new posts
if created and instance.published:
send_notification_email(instance)
# Update search index (if using search)
if hasattr(instance, 'update_search_index'):
instance.update_search_index()
def send_notification_email(post):
"""Send email notification for new post"""
try:
subject = f"New post: {post.title}"
message = f"A new post '{post.title}' has been published by {post.author.get_full_name()}"
# Get subscribers (implement your own logic)
subscribers = get_post_subscribers(post)
if subscribers:
send_mail(
subject=subject,
message=message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=subscribers,
fail_silently=True
)
except Exception as e:
logger.error(f"Failed to send notification email: {e}")
@receiver(pre_delete, sender=Post)
def post_pre_delete_handler(sender, instance, **kwargs):
"""Handle post deletion"""
# Log the deletion
logger.warning(f"Post '{instance.title}' is being deleted by user action")
# Clean up related files
if instance.featured_image:
instance.featured_image.delete(save=False)
# Clear cache
cache_key = f"post_{instance.id}"
cache.delete(cache_key)
cache.delete("recent_posts")
@receiver(m2m_changed, sender=Post.tags.through)
def post_tags_changed_handler(sender, instance, action, pk_set, **kwargs):
"""Handle changes to post tags"""
if action in ['post_add', 'post_remove', 'post_clear']:
# Clear tag-related cache
cache.delete(f"post_{instance.id}_tags")
# Update tag counts
if pk_set:
for tag_id in pk_set:
try:
tag = Tag.objects.get(id=tag_id)
update_tag_post_count(tag)
except Tag.DoesNotExist:
pass
def update_tag_post_count(tag):
"""Update the post count for a tag"""
count = tag.post_set.filter(published=True).count()
cache.set(f"tag_{tag.id}_count", count, 3600) # Cache for 1 hour
# Conditional signal handling
@receiver(post_save, sender=User)
def user_save_handler(sender, instance, created, **kwargs):
"""Handle user save with conditions"""
# Only for new users
if created:
# Send welcome email
send_welcome_email(instance)
# Create default preferences
UserPreferences.objects.create(
user=instance,
email_notifications=True,
theme='light'
)
# Only for existing users
else:
# Check if email changed
if 'update_fields' in kwargs:
update_fields = kwargs['update_fields']
if update_fields and 'email' in update_fields:
# Email was updated
send_email_change_notification(instance)
@receiver(post_save, sender=Post)
def post_save_conditional_handler(sender, instance, created, **kwargs):
"""Conditional post save handling"""
# Only handle published posts
if not instance.published:
return
# Skip if this is a bulk update
if kwargs.get('update_fields') is not None:
update_fields = kwargs['update_fields']
# Only proceed if specific fields were updated
if not any(field in update_fields for field in ['title', 'content', 'published']):
return
# Your signal logic here
update_post_cache(instance)
# Signal with sender filtering
@receiver(post_save)
def generic_model_save_handler(sender, instance, **kwargs):
"""Handle save for multiple model types"""
# Handle different model types
if sender == Post:
handle_post_save(instance, **kwargs)
elif sender == Comment:
handle_comment_save(instance, **kwargs)
elif sender == User:
handle_user_save(instance, **kwargs)
# apps.py - Proper signal registration
from django.apps import AppConfig
class BlogConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'blog'
def ready(self):
"""Import signals when app is ready"""
import blog.signals # noqa
# __init__.py in your app
default_app_config = 'blog.apps.BlogConfig'
# Alternative: Import in models.py
# models.py
from django.db import models
# ... your models ...
# Import signals at the end of models.py
from . import signals # noqa
# signals.py - Custom signal definitions
import django.dispatch
from django.dispatch import Signal
# Define custom signals
user_profile_updated = django.dispatch.Signal()
post_published = django.dispatch.Signal()
comment_approved = django.dispatch.Signal()
# Signals with providing_args (deprecated in Django 3.1+)
# Use type hints and documentation instead
order_completed = Signal() # providing_args=['order', 'user', 'total']
# Modern approach with documentation
class CustomSignals:
"""Custom signals for the application
Signals:
user_profile_updated: Sent when user profile is updated
Arguments: user, profile, changed_fields
post_published: Sent when a post is published
Arguments: post, author, publish_date
comment_approved: Sent when a comment is approved
Arguments: comment, moderator, approval_date
"""
user_profile_updated = Signal()
post_published = Signal()
comment_approved = Signal()
# Usage in models or views
from .signals import CustomSignals
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
published = models.BooleanField(default=False)
def publish(self):
"""Publish the post and send signal"""
if not self.published:
self.published = True
self.save()
# Send custom signal
CustomSignals.post_published.send(
sender=self.__class__,
post=self,
author=self.author,
publish_date=timezone.now()
)
class UserProfile(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
bio = models.TextField()
def save(self, *args, **kwargs):
"""Override save to track changes"""
changed_fields = []
if self.pk: # Existing instance
old_instance = UserProfile.objects.get(pk=self.pk)
# Track changed fields
for field in self._meta.fields:
field_name = field.name
old_value = getattr(old_instance, field_name)
new_value = getattr(self, field_name)
if old_value != new_value:
changed_fields.append(field_name)
super().save(*args, **kwargs)
# Send signal with changed fields
if changed_fields:
CustomSignals.user_profile_updated.send(
sender=self.__class__,
user=self.user,
profile=self,
changed_fields=changed_fields
)
# views.py - Sending signals from views
from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from .signals import CustomSignals
from .models import Post, Comment
@login_required
def approve_comment(request, comment_id):
"""Approve a comment and send signal"""
comment = get_object_or_404(Comment, id=comment_id)
if not request.user.has_perm('blog.change_comment'):
messages.error(request, "You don't have permission to approve comments")
return redirect('comment_list')
# Approve the comment
comment.approved = True
comment.approved_by = request.user
comment.approved_at = timezone.now()
comment.save()
# Send custom signal
CustomSignals.comment_approved.send(
sender=Comment,
comment=comment,
moderator=request.user,
approval_date=comment.approved_at
)
messages.success(request, f"Comment by {comment.author.username} has been approved")
return redirect('comment_list')
# services.py - Sending signals from service layer
class PostService:
"""Service class for post operations"""
@staticmethod
def bulk_publish_posts(post_ids, user):
"""Bulk publish posts and send signals"""
posts = Post.objects.filter(id__in=post_ids, published=False)
for post in posts:
post.published = True
post.published_by = user
post.published_at = timezone.now()
post.save()
# Send signal for each post
CustomSignals.post_published.send(
sender=Post,
post=post,
author=post.author,
publish_date=post.published_at
)
return posts.count()
@staticmethod
def feature_post(post_id, user):
"""Feature a post"""
post = Post.objects.get(id=post_id)
post.featured = True
post.featured_by = user
post.featured_at = timezone.now()
post.save()
# Send custom signal
post_featured = Signal()
post_featured.send(
sender=Post,
post=post,
featured_by=user,
featured_at=post.featured_at
)
# management/commands/publish_scheduled_posts.py
from django.core.management.base import BaseCommand
from django.utils import timezone
from blog.models import Post
from blog.signals import CustomSignals
class Command(BaseCommand):
help = 'Publish scheduled posts'
def handle(self, *args, **options):
"""Publish posts scheduled for now"""
now = timezone.now()
scheduled_posts = Post.objects.filter(
published=False,
scheduled_publish_date__lte=now
)
for post in scheduled_posts:
post.published = True
post.save()
# Send signal
CustomSignals.post_published.send(
sender=Post,
post=post,
author=post.author,
publish_date=now
)
self.stdout.write(
self.style.SUCCESS(f'Published post: {post.title}')
)
# Signal handlers for custom signals
from .signals import CustomSignals
from django.dispatch import receiver
from django.core.mail import send_mail
from django.template.loader import render_to_string
@receiver(CustomSignals.post_published)
def handle_post_published(sender, post, author, publish_date, **kwargs):
"""Handle post publication"""
# Update author statistics
author_profile = author.userprofile
author_profile.published_posts_count = author.posts.filter(published=True).count()
author_profile.save()
# Send notification to subscribers
subscribers = get_post_subscribers(author)
if subscribers:
send_post_notification(post, subscribers)
# Update sitemap
update_sitemap()
# Clear relevant caches
cache.delete_many([
'recent_posts',
f'author_{author.id}_posts',
'published_posts_count'
])
@receiver(CustomSignals.comment_approved)
def handle_comment_approved(sender, comment, moderator, approval_date, **kwargs):
"""Handle comment approval"""
# Notify comment author
if comment.author.email:
send_mail(
subject='Your comment has been approved',
message=f'Your comment on "{comment.post.title}" has been approved.',
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=[comment.author.email],
fail_silently=True
)
# Notify post author
if comment.post.author != comment.author and comment.post.author.email:
send_mail(
subject=f'New comment on your post: {comment.post.title}',
message=f'{comment.author.username} commented on your post.',
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=[comment.post.author.email],
fail_silently=True
)
@receiver(CustomSignals.user_profile_updated)
def handle_profile_updated(sender, user, profile, changed_fields, **kwargs):
"""Handle profile updates"""
# Log profile changes
logger.info(f"User {user.username} updated profile fields: {changed_fields}")
# Clear user cache
cache.delete(f'user_profile_{user.id}')
# If avatar changed, clear image cache
if 'avatar' in changed_fields:
cache.delete(f'user_avatar_{user.id}')
# Update search index if bio changed
if 'bio' in changed_fields:
update_user_search_index(user)
# Disconnecting signals
from django.db.models.signals import post_save
from django.dispatch import receiver
# Method 1: Disconnect specific handler
def my_handler(sender, **kwargs):
pass
# Connect
post_save.connect(my_handler, sender=User)
# Disconnect
post_save.disconnect(my_handler, sender=User)
# Method 2: Disconnect by receiver
@receiver(post_save, sender=User)
def user_save_handler(sender, **kwargs):
pass
# Disconnect
post_save.disconnect(user_save_handler, sender=User)
# Method 3: Disconnect all handlers for a sender
post_save.disconnect(sender=User)
# Temporary signal disconnection
from contextlib import contextmanager
from django.db.models.signals import post_save
@contextmanager
def disable_signals():
"""Context manager to temporarily disable signals"""
# Store original handlers
original_handlers = {}
signals_to_disable = [
(post_save, User, create_user_profile),
(post_save, Post, post_saved_handler),
(pre_delete, Post, post_pre_delete_handler),
]
# Disconnect signals
for signal, sender, handler in signals_to_disable:
signal.disconnect(handler, sender=sender)
try:
yield
finally:
# Reconnect signals
for signal, sender, handler in signals_to_disable:
signal.connect(handler, sender=sender)
# Usage
with disable_signals():
# Bulk operations without triggering signals
User.objects.bulk_create([
User(username=f'user{i}', email=f'user{i}@example.com')
for i in range(1000)
])
# Selective signal disconnection
class SignalManager:
"""Manage signal connections"""
def __init__(self):
self.disabled_signals = set()
def disable_signal(self, signal, sender, handler):
"""Disable a specific signal"""
key = (signal, sender, handler)
if key not in self.disabled_signals:
signal.disconnect(handler, sender=sender)
self.disabled_signals.add(key)
def enable_signal(self, signal, sender, handler):
"""Re-enable a specific signal"""
key = (signal, sender, handler)
if key in self.disabled_signals:
signal.connect(handler, sender=sender)
self.disabled_signals.remove(key)
def enable_all(self):
"""Re-enable all disabled signals"""
for signal, sender, handler in self.disabled_signals.copy():
signal.connect(handler, sender=sender)
self.disabled_signals.remove((signal, sender, handler))
# Usage
signal_manager = SignalManager()
# Disable specific signal
signal_manager.disable_signal(post_save, User, create_user_profile)
# Perform operations
User.objects.create(username='test', email='test@example.com')
# Re-enable
signal_manager.enable_signal(post_save, User, create_user_profile)
# test_signals.py
from django.test import TestCase, override_settings
from django.db.models.signals import post_save
from django.dispatch import receiver
from unittest.mock import patch, Mock
from .models import Post, User
from .signals import post_saved_handler
class SignalTestCase(TestCase):
"""Test signal functionality"""
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com'
)
def test_post_save_signal_triggered(self):
"""Test that post_save signal is triggered"""
# Mock the signal handler
with patch('blog.signals.post_saved_handler') as mock_handler:
# Create a post
post = Post.objects.create(
title='Test Post',
content='Test content',
author=self.user
)
# Verify signal was called
mock_handler.assert_called_once()
# Check call arguments
args, kwargs = mock_handler.call_args
self.assertEqual(kwargs['sender'], Post)
self.assertEqual(kwargs['instance'], post)
self.assertTrue(kwargs['created'])
def test_signal_with_mock_receiver(self):
"""Test signals with mock receiver"""
# Create mock receiver
mock_receiver = Mock()
# Connect mock receiver
post_save.connect(mock_receiver, sender=Post)
try:
# Create post
post = Post.objects.create(
title='Test Post',
content='Test content',
author=self.user
)
# Verify mock was called
mock_receiver.assert_called_once()
# Check arguments
call_kwargs = mock_receiver.call_args[1]
self.assertEqual(call_kwargs['sender'], Post)
self.assertEqual(call_kwargs['instance'], post)
finally:
# Clean up - disconnect mock receiver
post_save.disconnect(mock_receiver, sender=Post)
def test_signal_disconnection(self):
"""Test signal disconnection"""
# Disconnect the signal
post_save.disconnect(post_saved_handler, sender=Post)
try:
with patch('blog.signals.logger') as mock_logger:
# Create post
Post.objects.create(
title='Test Post',
content='Test content',
author=self.user
)
# Verify handler was not called
mock_logger.info.assert_not_called()
finally:
# Reconnect the signal
post_save.connect(post_saved_handler, sender=Post)
@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
def test_async_signal_handling(self):
"""Test asynchronous signal handling"""
with patch('blog.tasks.send_notification_email.delay') as mock_task:
# Create published post
post = Post.objects.create(
title='Test Post',
content='Test content',
author=self.user,
published=True
)
# Verify async task was called
mock_task.assert_called_once_with(post.id)
class CustomSignalTestCase(TestCase):
"""Test custom signals"""
def test_custom_signal_sending(self):
"""Test sending custom signals"""
from .signals import CustomSignals
# Mock receiver
mock_receiver = Mock()
CustomSignals.post_published.connect(mock_receiver)
try:
# Send signal
CustomSignals.post_published.send(
sender=Post,
post=Mock(id=1, title='Test'),
author=self.user,
publish_date=timezone.now()
)
# Verify receiver was called
mock_receiver.assert_called_once()
finally:
CustomSignals.post_published.disconnect(mock_receiver)
def test_signal_with_no_receivers(self):
"""Test sending signal with no receivers"""
from .signals import CustomSignals
# This should not raise an exception
responses = CustomSignals.post_published.send(
sender=Post,
post=Mock(id=1),
author=self.user,
publish_date=timezone.now()
)
# Should return empty list
self.assertEqual(responses, [])
# Performance optimizations for signals
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.cache import cache
import logging
logger = logging.getLogger(__name__)
@receiver(post_save, sender=Post)
def optimized_post_handler(sender, instance, created, **kwargs):
"""Optimized signal handler"""
# Early return for non-published posts
if not instance.published:
return
# Batch operations to reduce database hits
operations = []
if created:
operations.extend([
'send_notification',
'update_author_stats',
'clear_cache'
])
else:
# Only clear cache for updates
operations.append('clear_cache')
# Execute operations
for operation in operations:
try:
if operation == 'send_notification':
# Use async task for email sending
from .tasks import send_post_notification
send_post_notification.delay(instance.id)
elif operation == 'update_author_stats':
# Use F expressions to avoid race conditions
from django.db.models import F
UserProfile.objects.filter(user=instance.author).update(
post_count=F('post_count') + 1
)
elif operation == 'clear_cache':
# Batch cache deletion
cache_keys = [
f'post_{instance.id}',
'recent_posts',
f'author_{instance.author.id}_posts'
]
cache.delete_many(cache_keys)
except Exception as e:
logger.error(f"Error in signal operation {operation}: {e}")
# Async signal handling with Celery
from celery import shared_task
@shared_task
def async_post_save_handler(post_id, created):
"""Async signal handler"""
try:
post = Post.objects.select_related('author').get(id=post_id)
if created and post.published:
# Send notifications
send_notification_emails(post)
# Update search index
update_search_index(post)
# Generate social media posts
create_social_media_posts(post)
except Post.DoesNotExist:
logger.error(f"Post {post_id} not found in async handler")
except Exception as e:
logger.error(f"Error in async post handler: {e}")
@receiver(post_save, sender=Post)
def async_post_save_trigger(sender, instance, created, **kwargs):
"""Trigger async signal handling"""
# Only for published posts
if instance.published:
async_post_save_handler.delay(instance.id, created)
Django signals provide a powerful way to decouple your application logic and respond to events throughout your application. They're particularly useful for cross-cutting concerns like logging, caching, notifications, and maintaining data consistency. However, use them judiciously as they can make code harder to debug and test if overused.
Fixtures
Fixtures provide a way to pre-populate your database with data for testing, development, and initial application setup. Understanding how to create, manage, and use fixtures effectively enables consistent data management across different environments.
Migrations
Django migrations provide a version control system for your database schema, allowing you to evolve your models over time while maintaining data integrity. Understanding migrations is essential for managing database changes in development, testing, and production environments.