post_save, pre_delete, m2m_changed, custom signals, best practices
Signals позволяют коду реагировать на определённые события в приложении. Это механизм для развязки компонентов через события.
💡 Правило: Избегайте сигналов для бизнес-логики в одном приложении. Используйте для интеграции между независимыми приложениями или для кэш-инвалидации.
| Сигнал | Когда отправляется | Параметры |
|---|---|---|
pre_save | Перед вызовом save() | sender, instance, raw, using, update_fields |
post_save | После вызова save() | sender, instance, created, raw, using, update_fields |
pre_delete | Перед вызовом delete() | sender, instance, using |
post_delete | После вызова delete() | sender, instance, using |
m2m_changed | При изменении ManyToMany | sender, instance, action, reverse, model, pk_set, using |
class_prepared | После регистрации модели | sender, class |
from django.db.models.signals import pre_save, post_save, pre_delete, post_delete, m2m_changed
from django.core.signals import request_finished, request_started, got_request_exception
from django.dispatch import Signal# blog/signals.py
from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from django.core.mail import send_mail
from .models import Post
@receiver(post_save, sender=Post)
def notify_author_on_publish(sender, instance, created, **kwargs):
"""Отправить email автору при публикации поста."""
if created and instance.is_published:
send_mail(
subject='Your post is published!',
message=f'Your post "{instance.title}" is now live.',
from_email='noreply@example.com',
recipient_list=[instance.author.email],
fail_silently=True,
)
@receiver(pre_delete, sender=Post)
def backup_post_before_delete(sender, instance, **kwargs):
"""Создать резервную копию перед удалением."""
PostBackup.objects.create(
original_id=instance.id,
title=instance.title,
content=instance.content,
author=instance.author,
deleted_at=timezone.now()
)# blog/apps.py
from django.apps import AppConfig
class BlogConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'blog'
def ready(self):
# Импортируем signals для регистрации receivers
import blog.signals # noqa# settings.py
INSTALLED_APPS = [
# ...
'blog.apps.BlogConfig', # Или просто 'blog'
]# accounts/signals.py
from django.contrib.auth import get_user_model
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Profile
User = get_user_model()
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
"""Создать профиль при создании пользователя."""
if created:
Profile.objects.create(user=instance)
@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
"""Сохранить профиль при сохранении пользователя."""
instance.profile.save()# blog/signals.py
from django.core.cache import cache
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import Post
@receiver([post_save, post_delete], sender=Post)
def invalidate_post_cache(sender, instance, **kwargs):
"""Инвалидировать кэш при изменении поста."""
# Удалить кэш конкретного поста
cache.delete(f'post:{instance.id}')
# Удалить кэш списка постов
cache.delete('posts:list')
# Удалить кэш автора
cache.delete(f'author:{instance.author_id}:posts')# blog/signals.py
import json
from django.db.models.signals import pre_save, post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist
from .models import Post, PostRevision
User = get_user_model()
@receiver(pre_save, sender=Post)
def track_changes(sender, instance, **kwargs):
"""Отследить изменения перед сохранением."""
if not instance.pk:
# Новый объект
return
try:
old_instance = Post.objects.get(pk=instance.pk)
except ObjectDoesNotExist:
return
# Сохранить старые значения для последующего сравнения
instance._old_content = old_instance.content
instance._old_title = old_instance.title
instance._old_status = old_instance.status
@receiver(post_save, sender=Post)
def create_revision(sender, instance, created, **kwargs):
"""Создать ревизию при изменении."""
if created:
return
if not hasattr(instance, '_old_content'):
return
changes = {}
if instance._old_title != instance.title:
changes['title'] = {
'old': instance._old_title,
'new': instance.title
}
if instance._old_content != instance.content:
changes['content'] = {
'old': instance._old_content,
'new': instance.content
}
if instance._old_status != instance.status:
changes['status'] = {
'old': instance._old_status,
'new': instance.status
}
if changes:
PostRevision.objects.create(
post=instance,
changes=changes,
changed_by=get_current_user() # Нужно реализовать
)# blog/signals.py
from django.db.models.signals import m2m_changed
from django.dispatch import receiver
from .models import Post, Tag
@receiver(m2m_changed, sender=Post.tags.through)
def update_tag_post_count(sender, instance, action, pk_set, **kwargs):
"""Обновить счётчик постов у тегов."""
if action == 'post_add':
Tag.objects.filter(id__in=pk_set).update(
post_count=F('post_count') + 1
)
elif action == 'post_remove':
Tag.objects.filter(id__in=pk_set).update(
post_count=F('post_count') - 1
)
elif action == 'post_clear':
# Все теги удалены
old_tags = instance.tags.all()
old_tags.update(post_count=F('post_count') - 1)# tasks/signals.py
from django.dispatch import Signal
# Определение сигналов
task_started = Signal()
task_completed = Signal()
task_failed = Signal()
# С параметрами
task_completed = Signal(providing_args=['task_id', 'result', 'duration'])# tasks/models.py
from django.db import models
from django.dispatch import Signal
from .signals import task_started, task_completed, task_failed
class Task(models.Model):
name = models.CharField(max_length=200)
status = models.CharField(max_length=20, default='pending')
result = models.TextField(null=True, blank=True)
def execute(self):
"""Выполнить задачу с отправкой сигналов."""
task_started.send(sender=self, task_id=self.id)
try:
# Выполнение задачи
self.status = 'running'
self.save()
result = self._do_work()
self.status = 'completed'
self.result = result
self.save()
task_completed.send(
sender=self,
task_id=self.id,
result=result,
duration=1.5
)
except Exception as e:
self.status = 'failed'
self.result = str(e)
self.save()
task_failed.send(
sender=self,
task_id=self.id,
error=str(e)
)
def _do_work(self):
# Логика выполнения
return 'Done'# tasks/signals_handlers.py
from django.dispatch import receiver
from django.core.mail import send_mail
from .signals import task_started, task_completed, task_failed
@receiver(task_started)
def log_task_start(sender, task_id, **kwargs):
"""Логировать начало задачи."""
print(f'Task {task_id} started')
@receiver(task_completed)
def notify_task_complete(sender, task_id, result, **kwargs):
"""Уведомить о завершении."""
send_mail(
subject='Task completed',
message=f'Task {task_id} completed with result: {result}',
from_email='noreply@example.com',
recipient_list=['admin@example.com'],
)
@receiver(task_failed)
def notify_task_failed(sender, task_id, error, **kwargs):
"""Уведомить об ошибке."""
send_mail(
subject='Task failed',
message=f'Task {task_id} failed: {error}',
from_email='noreply@example.com',
recipient_list=['admin@example.com'],
)✅ Хорошие сценарии:
❌ Плохие сценарии:
# ❌ Проблема: неявный поток выполнения
@receiver(post_save, sender=Post)
def do_something(sender, instance, **kwargs):
# Кто знает что здесь происходит?
pass
@receiver(post_save, sender=Post)
def do_something_else(sender, instance, **kwargs):
# И здесь тоже
pass
# ✅ Решение: явные вызовы
class PostService:
def create_post(self, **data):
post = Post.objects.create(**data)
self.notify_author(post)
self.invalidate_cache(post)
self.log_creation(post)
return post# ❌ Проблема: дублирование при bulk операциях
Post.objects.bulk_create(posts) # Сигналы НЕ вызываются!
# ✅ Решение: явный цикл
for post in posts:
post.save() # Сигналы вызываются
# ✅ Или переписать без сигналов
def create_posts(posts):
Post.objects.bulk_create(posts)
invalidate_cache() # Явная инвалидация# tests.py
from django.test import TestCase
from django.db.models.signals import post_save
from .signals import my_handler
from .models import Post
class PostTest(TestCase):
def setUp(self):
# Отключить сигнал для тестов
post_save.disconnect(my_handler, sender=Post)
def tearDown(self):
# Включить обратно
post_save.connect(my_handler, sender=Post)
def test_post_creation(self):
# Тест без вызова сигнала
post = Post.objects.create(title='Test')
self.assertEqual(post.title, 'Test')# audit/signals.py
from django.db.models.signals import post_save, post_delete, m2m_changed
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist
from django.utils import timezone
import json
User = get_user_model()
class AuditLog:
"""Класс для логирования изменений."""
@staticmethod
def log_action(model, instance, action, user=None, changes=None):
"""Записать действие в лог."""
AuditEntry.objects.create(
model_name=model.__name__,
object_id=instance.pk if instance else None,
action=action,
user=user,
changes=changes or {},
timestamp=timezone.now()
)
@staticmethod
def get_current_user():
"""Получить текущего пользователя (требует middleware)."""
from audit.middleware import get_current_user
return get_current_user()
@receiver(post_save)
def audit_post_save(sender, instance, created, **kwargs):
"""Логировать создание и обновление."""
# Игнорировать системные модели
if sender._meta.app_label in ['auth', 'contenttypes', 'sessions']:
return
action = 'create' if created else 'update'
changes = {}
if not created and hasattr(instance, '_old_values'):
old_values = instance._old_values
for field_name, old_value in old_values.items():
new_value = getattr(instance, field_name)
if old_value != new_value:
changes[field_name] = {
'old': str(old_value),
'new': str(new_value)
}
AuditLog.log_action(
model=sender,
instance=instance,
action=action,
user=AuditLog.get_current_user(),
changes=changes
)
@receiver(post_delete)
def audit_post_delete(sender, instance, **kwargs):
"""Логировать удаление."""
if sender._meta.app_label in ['auth', 'contenttypes', 'sessions']:
return
AuditLog.log_action(
model=sender,
instance=instance,
action='delete',
user=AuditLog.get_current_user()
)# audit/models.py
from django.db import models
from django.contrib.auth import get_user_model
User = get_user_model()
class AuditEntry(models.Model):
"""Запись аудита изменений."""
ACTION_CHOICES = [
('create', 'Create'),
('update', 'Update'),
('delete', 'Delete'),
]
model_name = models.CharField(max_length=255)
object_id = models.IntegerField(null=True, blank=True)
action = models.CharField(max_length=20, choices=ACTION_CHOICES)
user = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
blank=True
)
changes = models.JSONField(default=dict, blank=True)
timestamp = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-timestamp']
indexes = [
models.Index(fields=['model_name', 'object_id']),
models.Index(fields=['action']),
models.Index(fields=['-timestamp']),
]
def __str__(self):
return f'{self.action} {self.model_name} ({self.object_id}) at {self.timestamp}'# audit/middleware.py
"""Middleware для получения текущего пользователя в signals."""
import threading
_thread_locals = threading.local()
class AuditUserMiddleware:
"""Сохраняет текущего пользователя в thread-local."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
_thread_locals.current_user = getattr(request, 'user', None)
response = self.get_response(request)
return response
def get_current_user():
"""Получить текущего пользователя из thread-local."""
return getattr(_thread_locals, 'current_user', None)Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.