Django Signals: A Comprehensive Guide
Introduction
Django signals are a way to allow decoupled applications to get notified when certain actions occur elsewhere in the application. They provide a way to send notifications when certain actions occur, allowing different parts of your application to respond to these events without being directly coupled to each other.
What are Signals?
Signals are Django’s implementation of the observer pattern. They allow certain senders to notify a set of receivers when some action has taken place. Signals are especially useful when many pieces of code may be interested in the same events.
Key Concepts
- Sender: The object that sends the signal
- Receiver: The function that receives the signal
- Signal: The notification mechanism itself
- Dispatch: The process of sending signals to receivers
Signal Architecture
Action Occurs → Signal Sent → Receivers Notified → Receivers ExecuteSignal Flow
- Action: Something happens in your application (e.g., a model is saved)
- Signal: A signal is dispatched with relevant data
- Receivers: Functions registered to listen for that signal are called
- Execution: Each receiver function executes with the signal data
Built-in Django Signals
Model Signals
1. pre_save / post_save
from django.db.models.signals import pre_save, post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
# pre_save - called before a model's save() method is called
@receiver(pre_save, sender=User)
def pre_save_user(sender, instance, **kwargs):
print(f"About to save user: {instance.username}")
# Modify instance before saving
if not instance.username:
instance.username = instance.email.split('@')[0]
# post_save - called after a model's save() method is called
@receiver(post_save, sender=User)
def post_save_user(sender, instance, created, **kwargs):
if created:
print(f"New user created: {instance.username}")
# Send welcome email, create profile, etc.
else:
print(f"User updated: {instance.username}")2. pre_delete / post_delete
from django.db.models.signals import pre_delete, post_delete
from django.dispatch import receiver
from myapp.models import Article
@receiver(pre_delete, sender=Article)
def pre_delete_article(sender, instance, **kwargs):
print(f"About to delete article: {instance.title}")
# Backup data, notify users, etc.
@receiver(post_delete, sender=Article)
def post_delete_article(sender, instance, **kwargs):
print(f"Article deleted: {instance.title}")
# Clean up related files, update cache, etc.3. m2m_changed
from django.db.models.signals import m2m_changed
from django.dispatch import receiver
from myapp.models import Article, Tag
@receiver(m2m_changed, sender=Article.tags.through)
def m2m_changed_article_tags(sender, instance, action, pk_set, **kwargs):
if action == "post_add":
print(f"Tags added to article {instance.title}: {pk_set}")
elif action == "post_remove":
print(f"Tags removed from article {instance.title}: {pk_set}")
elif action == "post_clear":
print(f"All tags cleared from article {instance.title}")Request/Response Signals
1. request_started / request_finished
from django.core.signals import request_started, request_finished
from django.dispatch import receiver
@receiver(request_started)
def request_started_handler(sender, **kwargs):
print("Request started")
@receiver(request_finished)
def request_finished_handler(sender, **kwargs):
print("Request finished")2. got_request_exception
from django.core.signals import got_request_exception
from django.dispatch import receiver
import logging
logger = logging.getLogger(__name__)
@receiver(got_request_exception)
def exception_handler(sender, request, **kwargs):
logger.error(f"Exception in request: {request.path}")
# Send notification, log to external service, etc.Database Signals
1. connection_created
from django.db.backends.signals import connection_created
from django.dispatch import receiver
@receiver(connection_created)
def connection_created_handler(sender, connection, **kwargs):
print(f"Database connection created: {connection.settings_dict['NAME']}")Creating Custom Signals
Defining Custom Signals
# signals.py
from django.dispatch import Signal
# Define custom signals
user_registered = Signal()
order_placed = Signal()
payment_received = Signal()
article_published = Signal()
# Signals with arguments
user_registered = Signal(providing_args=["user", "created_at"])
order_placed = Signal(providing_args=["order", "user", "total"])
payment_received = Signal(providing_args=["payment", "amount", "currency"])Sending Custom Signals
# views.py
from django.dispatch import Signal
from .signals import user_registered, order_placed, payment_received
def register_user(request):
# Create user logic
user = User.objects.create_user(username='john', email='john@example.com')
# Send signal
user_registered.send(
sender=User,
user=user,
created_at=timezone.now()
)
return HttpResponse("User registered")
def place_order(request):
# Order creation logic
order = Order.objects.create(user=request.user, total=100.00)
# Send signal
order_placed.send(
sender=Order,
order=order,
user=request.user,
total=order.total
)
return HttpResponse("Order placed")Receiving Custom Signals
# receivers.py
from django.dispatch import receiver
from .signals import user_registered, order_placed, payment_received
@receiver(user_registered)
def handle_user_registration(sender, user, created_at, **kwargs):
print(f"New user registered: {user.username}")
# Send welcome email
send_welcome_email(user)
# Create user profile
create_user_profile(user)
@receiver(order_placed)
def handle_order_placed(sender, order, user, total, **kwargs):
print(f"Order placed by {user.username}: ${total}")
# Send order confirmation
send_order_confirmation(order)
# Update inventory
update_inventory(order)
@receiver(payment_received)
def handle_payment_received(sender, payment, amount, currency, **kwargs):
print(f"Payment received: {amount} {currency}")
# Send receipt
send_payment_receipt(payment)
# Update order status
update_order_status(payment.order)Advanced Signal Usage
Conditional Signal Handling
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import User, Profile
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
if created: # Only for new users
Profile.objects.create(user=instance)
print(f"Profile created for {instance.username}")
@receiver(post_save, sender=User)
def update_user_profile(sender, instance, created, **kwargs):
if not created: # Only for existing users
try:
profile = instance.profile
profile.last_updated = timezone.now()
profile.save()
print(f"Profile updated for {instance.username}")
except Profile.DoesNotExist:
Profile.objects.create(user=instance)Signal with Multiple Receivers
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Order
@receiver(post_save, sender=Order)
def send_order_confirmation(sender, instance, created, **kwargs):
if created:
send_email_confirmation(instance)
@receiver(post_save, sender=Order)
def update_inventory(sender, instance, created, **kwargs):
if created:
update_product_stock(instance)
@receiver(post_save, sender=Order)
def notify_admin(sender, instance, created, **kwargs):
if created and instance.total > 1000:
notify_admin_of_large_order(instance)
@receiver(post_save, sender=Order)
def update_analytics(sender, instance, created, **kwargs):
update_sales_analytics(instance)Signal with Error Handling
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import User
import logging
logger = logging.getLogger(__name__)
@receiver(post_save, sender=User)
def handle_user_save(sender, instance, created, **kwargs):
try:
if created:
# Create user profile
create_user_profile(instance)
# Send welcome email
send_welcome_email(instance)
# Add to mailing list
add_to_mailing_list(instance)
else:
# Update user profile
update_user_profile(instance)
# Sync with external service
sync_with_external_service(instance)
except Exception as e:
logger.error(f"Error in user save signal: {str(e)}")
# Don't let signal errors break the save operationSignal with Async Processing
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.cache import cache
from myapp.models import Article
import threading
@receiver(post_save, sender=Article)
def handle_article_save(sender, instance, created, **kwargs):
# Clear cache immediately
cache.delete('recent_articles')
# Process heavy operations asynchronously
if created:
thread = threading.Thread(
target=process_new_article,
args=(instance.id,)
)
thread.start()
def process_new_article(article_id):
# Heavy processing that doesn't need to block the save
article = Article.objects.get(id=article_id)
generate_article_preview(article)
update_search_index(article)
notify_subscribers(article)Signal Best Practices
1. Import Signals in apps.py
# 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 when app is ready2. Use Signal Decorators
from django.dispatch import receiver
from django.db.models.signals import post_save
from myapp.models import User
@receiver(post_save, sender=User)
def user_post_save(sender, instance, created, **kwargs):
# Signal handler code
pass3. Avoid Circular Imports
# signals.py
from django.dispatch import Signal
# Define signals without importing models
user_registered = Signal(providing_args=["user"])
# receivers.py
from django.dispatch import receiver
from .signals import user_registered
@receiver(user_registered)
def handle_user_registration(sender, user, **kwargs):
# Import models here to avoid circular imports
from .models import Profile
Profile.objects.create(user=user)4. Use Weak References
from django.dispatch import receiver
from django.db.models.signals import post_save
from myapp.models import User
@receiver(post_save, sender=User, weak=True) # Use weak references
def user_post_save(sender, instance, created, **kwargs):
# Signal handler code
pass5. Handle Signal Errors Gracefully
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Order
import logging
logger = logging.getLogger(__name__)
@receiver(post_save, sender=Order)
def handle_order_save(sender, instance, created, **kwargs):
try:
if created:
send_order_confirmation(instance)
except Exception as e:
logger.error(f"Failed to send order confirmation: {str(e)}")
# Don't let signal errors break the save operationSignal Testing
Testing Signal Receivers
from django.test import TestCase
from django.contrib.auth.models import User
from myapp.models import Profile
from unittest.mock import patch
class SignalTestCase(TestCase):
def test_user_profile_created(self):
# Test that profile is created when user is created
user = User.objects.create_user(
username='testuser',
email='test@example.com'
)
# Check that profile was created
self.assertTrue(hasattr(user, 'profile'))
self.assertIsInstance(user.profile, Profile)
@patch('myapp.signals.send_welcome_email')
def test_welcome_email_sent(self, mock_send_email):
# Test that welcome email is sent
user = User.objects.create_user(
username='testuser',
email='test@example.com'
)
# Check that send_welcome_email was called
mock_send_email.assert_called_once_with(user)Testing Custom Signals
from django.test import TestCase
from django.dispatch import receiver
from myapp.signals import user_registered
from unittest.mock import Mock
class CustomSignalTestCase(TestCase):
def test_custom_signal(self):
# Create a mock receiver
mock_receiver = Mock()
# Connect the mock receiver to the signal
user_registered.connect(mock_receiver)
# Send the signal
user = User.objects.create_user(username='testuser')
user_registered.send(sender=User, user=user)
# Check that the receiver was called
mock_receiver.assert_called_once()
args, kwargs = mock_receiver.call_args
self.assertEqual(kwargs['user'], user)Performance Considerations
1. Avoid Expensive Operations in Signals
# Bad: Expensive operation in signal
@receiver(post_save, sender=User)
def expensive_operation(sender, instance, created, **kwargs):
if created:
# This blocks the save operation
heavy_processing(instance)
# Good: Use async processing
@receiver(post_save, sender=User)
def async_operation(sender, instance, created, **kwargs):
if created:
# Process asynchronously
threading.Thread(target=heavy_processing, args=(instance.id,)).start()2. Use Signal Caching
from django.core.cache import cache
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Article
@receiver(post_save, sender=Article)
def update_cache(sender, instance, created, **kwargs):
# Clear cache instead of recalculating
cache.delete('recent_articles')
cache.delete('article_count')3. Batch Signal Processing
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Order
from django.core.cache import cache
@receiver(post_save, sender=Order)
def batch_update_analytics(sender, instance, created, **kwargs):
# Instead of updating immediately, schedule batch update
cache.set('analytics_update_needed', True, 300) # 5 minutesCommon Signal Patterns
1. Model Lifecycle Management
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from myapp.models import User, UserProfile, UserSettings
@receiver(post_save, sender=User)
def create_user_related_objects(sender, instance, created, **kwargs):
if created:
UserProfile.objects.create(user=instance)
UserSettings.objects.create(user=instance)
@receiver(post_delete, sender=User)
def cleanup_user_related_objects(sender, instance, **kwargs):
# Clean up related objects when user is deleted
UserProfile.objects.filter(user=instance).delete()
UserSettings.objects.filter(user=instance).delete()2. Cache Management
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from myapp.models import Product, Category
@receiver([post_save, post_delete], sender=Product)
def invalidate_product_cache(sender, instance, **kwargs):
cache.delete(f'product_{instance.id}')
cache.delete('product_list')
cache.delete(f'category_products_{instance.category.id}')
@receiver([post_save, post_delete], sender=Category)
def invalidate_category_cache(sender, instance, **kwargs):
cache.delete(f'category_{instance.id}')
cache.delete('category_list')3. Notification System
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Comment, Notification
@receiver(post_save, sender=Comment)
def notify_comment_author(sender, instance, created, **kwargs):
if created:
# Notify post author about new comment
if instance.post.author != instance.author:
Notification.objects.create(
user=instance.post.author,
message=f"New comment on your post by {instance.author.username}",
related_object=instance
)Interview Questions and Answers
Q1: What are Django signals and how do they work?
A: Django signals are a way to allow decoupled applications to get notified when certain actions occur. They implement the observer pattern where senders notify receivers when events happen. Signals are useful for handling side effects without tightly coupling different parts of your application.
Q2: What are the main types of Django signals?
A: The main types are:
- Model signals: pre_save, post_save, pre_delete, post_delete, m2m_changed
- Request signals: request_started, request_finished, got_request_exception
- Database signals: connection_created
- Custom signals: User-defined signals for specific use cases
Q3: How do you create and use custom signals?
A:
# Define signal
from django.dispatch import Signal
user_registered = Signal(providing_args=["user"])
# Send signal
user_registered.send(sender=User, user=user)
# Receive signal
@receiver(user_registered)
def handle_registration(sender, user, **kwargs):
# Handle the signal
passQ4: What is the difference between pre_save and post_save signals?
A:
pre_saveis called before a model’s save() method is called, allowing you to modify the instance before it’s savedpost_saveis called after a model’s save() method is called, allowing you to perform actions after the save is complete
Q5: How do you handle signal errors?
A: Wrap signal handlers in try-except blocks to prevent signal errors from breaking the main operation:
@receiver(post_save, sender=User)
def handle_user_save(sender, instance, created, **kwargs):
try:
# Signal logic
pass
except Exception as e:
logger.error(f"Signal error: {str(e)}")Q6: What are the best practices for using signals?
A: Best practices include:
- Import signals in apps.py to ensure they’re loaded
- Use signal decorators for cleaner code
- Avoid circular imports by importing models inside signal handlers
- Handle errors gracefully
- Avoid expensive operations in signals
- Use weak references to prevent memory leaks
Q7: How do you test signals?
A: You can test signals by:
- Testing that signal receivers are called when expected
- Using mock objects to verify signal behavior
- Testing the side effects of signals
- Using Django’s test framework to create test scenarios
Q8: What are the performance implications of signals?
A: Signals can impact performance if:
- Expensive operations are performed synchronously
- Too many signals are sent frequently
- Signal handlers are not optimized
- Signals cause database queries in loops
Summary
Django signals provide a powerful way to decouple different parts of your application and handle side effects. They are useful for:
- Model Lifecycle Management: Creating related objects, cleaning up data
- Cache Management: Invalidating caches when data changes
- Notifications: Sending emails, creating notifications
- Audit Logging: Tracking changes to models
- Integration: Connecting with external services
Key points to remember:
- Signals are executed synchronously by default
- Use signals for side effects, not core business logic
- Handle signal errors gracefully
- Test your signal handlers thoroughly
- Consider performance implications
- Follow Django’s signal patterns and best practices
Signals are essential for building maintainable and decoupled Django applications.
Interview angle
- “When are signals appropriate?” - decoupling genuinely optional side effects, especially across apps you don’t control. For logic that must happen, an explicit call in a service is clearer and testable.
- “Why do signals get criticised?” - they make control flow invisible: a save triggers behaviour with no reference at the call site, which makes debugging and reasoning harder. They also fire inside the transaction, so side effects can occur for changes that later roll back.
- “How do you avoid the rollback problem?” -
transaction.on_commit()so the side effect runs only after a successful commit. Sending an email from apost_savesignal is the canonical bug this fixes.