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#
- ✅ Import in ready() - Always import signals in AppConfig.ready()
- ✅ Use @receiver - Cleaner than manual connect()
- ✅ Check created - Use
createdparameter in post_save - ✅ Avoid save() in signals - Can cause infinite loops
- ✅ Use update_fields - When saving in signals
Next: Celery & Celery Beat