- Что такое сигналы
- Встроенные сигналы
- Создание собственных сигналов
- Обработка сигналов
- Лучшие практики
- Примеры использования
Сигналы в Django - это механизм, который позволяет отдельным компонентам приложения уведомлять другие компоненты о происходящих событиях. Это реализация паттерна "Наблюдатель" (Observer), где одни части приложения могут "подписываться" на события, происходящие в других частях.
- Слабая связанность кода
- Асинхронное выполнение
- Возможность множественных обработчиков
- Централизованное управление событиями
Django предоставляет набор встроенных сигналов для различных событий:
from django.db.models.signals import (
pre_save,
post_save,
pre_delete,
post_delete,
m2m_changed
)
"""
pre_save - перед сохранением модели
post_save - после сохранения модели
pre_delete - перед удалением модели
post_delete - после удаления модели
m2m_changed - при изменении отношений many-to-many
"""
from django.core.signals import (
request_started,
request_finished,
got_request_exception
)
"""
request_started - когда Django начинает обработку запроса
request_finished - когда Django заканчивает обработку запроса
got_request_exception - когда возникает исключение при обработке запроса
"""
# signals.py
from django.dispatch import Signal
# Создание собственного сигнала
user_logged_in_failed = Signal() # Сигнал при неудачной попытке входа
order_completed = Signal() # Сигнал при завершении заказа
# apps.py
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = 'myapp'
def ready(self):
"""
Импортируем обработчики сигналов при загрузке приложения
"""
import myapp.signals.handlers
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.contrib.auth.models import User
@receiver(post_save, sender=User)
def handle_user_save(sender, instance, created, **kwargs):
"""
Обработчик сигнала сохранения пользователя
Args:
sender: Модель, отправившая сигнал
instance: Экземпляр модели
created: Boolean, True если создана новая запись
**kwargs: Дополнительные аргументы
"""
if created:
# Действия при создании нового пользователя
Profile.objects.create(user=instance)
from django.db.models.signals import post_save
def handle_order_save(sender, instance, created, **kwargs):
"""
Обработчик сигнала сохранения заказа
"""
if created:
# Отправка уведомления о новом заказе
send_order_notification(instance)
# Подключение обработчика к сигналу
post_save.connect(handle_order_save, sender=Order)
# Отправка сигнала
order_completed.send(
sender=self.__class__,
order=self,
user=self.user
)
# signals/
# __init__.py
# handlers.py
# custom_signals.py
# handlers.py
from django.dispatch import receiver
from django.db.models.signals import post_save
from .custom_signals import order_completed
@receiver(order_completed)
def handle_order_completion(sender, order, user, **kwargs):
"""
Централизованная обработка завершения заказа
"""
# Отправка email
send_order_confirmation_email(order, user)
# Обновление статистики
update_sales_statistics(order)
# Уведомление администратора
notify_admin_about_order(order)
from django.db.models.signals import post_save
from django.core.signals import request_finished
class DisableSignals:
"""
Контекстный менеджер для отключения сигналов
"""
def __init__(self, signals):
self.signals = signals
self.stashed_receivers = {}
def __enter__(self):
for signal in self.signals:
self.stashed_receivers[signal] = signal.receivers
signal.receivers = []
def __exit__(self, exc_type, exc_val, exc_tb):
for signal in self.signals:
signal.receivers = self.stashed_receivers[signal]
# Использование
with DisableSignals([post_save, request_finished]):
# Код, выполняемый без сигналов
user.save()
from django.contrib.auth.models import User
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Profile
@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()
from django.db.models.signals import pre_save
from django.dispatch import receiver
from .models import Article
import logging
logger = logging.getLogger(__name__)
@receiver(pre_save, sender=Article)
def log_article_changes(sender, instance, **kwargs):
"""
Логирование изменений в статье
"""
if instance.pk: # Если объект уже существует
old_instance = sender.objects.get(pk=instance.pk)
for field in ['title', 'content', 'author']:
old_value = getattr(old_instance, field)
new_value = getattr(instance, field)
if old_value != new_value:
logger.info(
f'Article {instance.pk} field "{field}" '
f'changed from "{old_value}" to "{new_value}"'
)
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from .services import PaymentService, ShippingService
@receiver(post_save, sender=Order)
def handle_order_processing(sender, instance, created, **kwargs):
"""
Обработка заказа во внешних сервисах
"""
if created:
# Создание платежа
payment_service = PaymentService()
payment_service.create_payment(instance)
# Создание доставки
shipping_service = ShippingService()
shipping_service.schedule_delivery(instance)
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from .models import BlogPost
@receiver([post_save, post_delete], sender=BlogPost)
def invalidate_blog_cache(sender, instance, **kwargs):
"""
Инвалидация кэша при изменении блога
"""
# Очистка кэша списка постов
cache.delete('blog_posts_list')
# Очистка кэша конкретного поста
cache.delete(f'blog_post_{instance.pk}')
# Очистка кэша связанных объектов
for tag in instance.tags.all():
cache.delete(f'tag_posts_{tag.pk}')