Привет, Хабр!

Сегодня я расскажу, как использовать Django Signals, чтобы приложение работало как часы. Signals — это встроенный механизм в Django, который позволяет разным частям приложения «общаться» друг с другом через события.

Зачем они нужны:

  • Нужно реагировать на определённые действия в приложении (например, сохранение объекта или завершение запроса).

  • Нужно изолировать логику обработки событий, чтобы основной код оставался чистым.

  • Не хотите вмешиваться в сторонние приложения, но хотите знать, когда там что‑то происходит.

Как это работает:

  1. Отправитель (Sender): генерирует событие.

  2. Сигнал (Signal): уведомляет зарегистрированные обработчики.

  3. Обработчики (Receivers): реагируют на сигнал, выполняя нужную логику.

Когда использовать Signals?

Когда они полезны:

  • Разделение логики. Уведомления, фоновые задачи, простая обработка событий — всё это отлично ложится на сигналы.

  • Модульность. Можно добавлять обработчики без изменения основного кода.

  • Многоуровневая обработка. Например, сохранение объекта запускает несколько независимых действий.

Когда лучше не использовать:

  • Прямая зависимость. Если обработчик используется только в одном месте, вызовите функцию напрямую.

  • Сложная бизнес‑логика. Переносить её в сигналы — плохая идея. Вы создадите код, который сложно понять и отладить.

Основы работы с Signals

Django имеет массу встроенных сигналов:

  • pre_save и post_save — до и после сохранения объекта.

  • pre_delete и post_delete — до и после удаления объекта.

  • request_started и request_finished — в начале и конце запроса.

  • user_logged_in, user_logged_out — пользователь вошёл/вышел.

Как подключать обработчики:

  1. Через метод connect:

    from django.db.models.signals import post_save
    from myapp.models import MyModel
    
    def my_handler(sender, **kwargs):
        print("Объект сохранён!")
    
    post_save.connect(my_handler, sender=MyModel)
  2. Через декоратор @receiver:

    from django.db.models.signals import post_save
    from django.dispatch import receiver
    from myapp.models import MyModel
    
    @receiver(post_save, sender=MyModel)
    def my_handler(sender, **kwargs):
        print("Объект сохранён!")

Пример применения на магазине котиков

Итак, у нас есть магазин котиков. Что мы хотим:

  1. Уведомлять пользователей о регистрации.

  2. Обрабатывать заказы.

  3. Уведомлять о поступлении новых котиков.

  4. Проверять зависимые данные при удалении котика.

Уведомление о регистрации

Когда пользователь регистрируется, отправляем ему приветственное письмо.

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.mail import send_mail
from myapp.models import CustomUser

@receiver(post_save, sender=CustomUser)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        send_mail(
            subject='Добро пожаловать в Магазин Котиков!',
            message=f'Привет, {instance.username}! Рады видеть тебя среди любителей котиков!',
            from_email='no-reply@catshop.com',
            recipient_list=[instance.email],
        )

Сигнал post_save ловит событие создания нового пользователя. Проверяем, что объект новый, и отправляем письмо.

Обработка заказа

Когда пользователь покупает котика, нужно:

  • Уменьшить его количество на складе.

  • Отправить уведомление о покупке.

  • Логировать заказ.

from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Order, Cat
from django.core.mail import send_mail

@receiver(post_save, sender=Order)
def process_order(sender, instance, created, **kwargs):
    if created:
        cat = instance.cat
        if cat.stock < instance.quantity:
            raise ValueError("Недостаточно котиков на складе!")
        cat.stock -= instance.quantity
        cat.save()

        send_mail(
            subject='Ваш заказ принят!',
            message=f'Спасибо за покупку котика {cat.name}! Скоро он будет у вас.',
            from_email='no-reply@catshop.com',
            recipient_list=[instance.user.email],
        )

        print(f"Пользователь {instance.user.username} купил {instance.quantity} котика(ов) {cat.name}.")

Уведомление о новых котиках

Когда поступают новые котики, отправляем уведомление подписчикам.

from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import Cat, Subscription
from django.core.mail import send_mail

@receiver(post_save, sender=Cat)
def notify_about_new_cats(sender, instance, created, **kwargs):
    if created and instance.stock > 0:
        subscribers = Subscription.objects.filter(is_active=True).select_related('user')
        emails = [sub.user.email for sub in subscribers]

        send_mail(
            subject='Новые котики на складе!',
            message=f'У нас появились новые котики: {instance.name}. Цена: {instance.price} руб. Успейте купить!',
            from_email='no-reply@catshop.com',
            recipient_list=emails,
        )

Удаление котика с проверкой зависимостей

Проверяем, нет ли активных заказов на удаляемого котика.

from django.db.models.signals import pre_delete
from django.dispatch import receiver
from myapp.models import Cat, Order
from django.core.mail import send_mail

@receiver(pre_delete, sender=Cat)
def check_before_delete(sender, instance, **kwargs):
    active_orders = Order.objects.filter(cat=instance)
    if active_orders.exists():
        raise ValueError(f"Котик {instance.name} фигурирует в заказах. Удаление невозможно.")

    send_mail(
        subject='Котик удалён из магазина',
        message=f'Котик {instance.name} был удалён. Проверьте связанные заказы.',
        from_email='no-reply@catshop.com',
        recipient_list=['manager@catshop.com'],
    )

Прочие фичи Signals

Доступна асинхронная обработка заказа для долгих задач:

@receiver(post_save, sender=Order)
async def async_process_order(sender, instance, **kwargs):
    await asyncio.sleep(1)  # Эмуляция долгой операции
    print(f"Асинхронная обработка заказа на {instance.cat.name}.")

Также можно создавать кастомные сигналы. Создадим свой сигнал «котик продан»:

from django.dispatch import Signal

cat_sold = Signal()

@receiver(cat_sold)
def notify_about_sale(sender, cat, user, **kwargs):
    print(f"Котик {cat.name} был продан пользователю {user.username}.")

Возможные ошибки и как их избежать

  1. Дублирование обработчиков: используйте dispatch_uid.

  2. Пропущенные аргументы: всегда добавляйте sender и **kwargs.

  3. Избыточность сигналов: используйте их только для изоляции логики.

  4. Пропуск sender: всегда указывайте конкретный отправитель.

  5. Неправильное место хранения: храните сигналы в signals.py и подключайте через apps.py.


А как вы используете Signals? Делитесь своими кейсами в комментариях! ?

Python-разработчикам, заинтересованным в профессиональном развитии, рекомендую обратить внимание на открытые уроки:

  • 13 января. Docker для Python-разработчика: разберём лучшие практики написания Dockerfile, изучим принципы работы с Docker и обсудим тонкости контейнеризации приложений. Особое внимание уделим специфике контейнеризации Python-приложений. Записаться

  • 23 января. Асинхронное взаимодействие в Python на примере RabbitMQ: рассмотрим пример построения архитектуры приложения, разберемся в преимуществах и недостатках такого подхода. Записаться

Комментарии (1)


  1. GospodinKolhoznik
    13.01.2025 14:09

    А как вы используете Signals?

    Безотносительно Django, как только приходится работать с каким либо фрэймворком, работающим через signals-ы, первым делом пишу API-обёртку, позволяющую избавиться от них.