Какие знаешь ограничения у сигналов в Django?

Какие знаю ограничения у сигналов в Django

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

Основные ограничения

1. Синхронность сигналов

Все сигналы выполняются синхронно в том же потоке, что и основной код. Если в обработчике долгая операция, это заблокирует всю систему:

from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import User
import time

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        time.sleep(5)  # ПЛОХО: блокирует основной поток
        send_email(instance.email)
        # Пользователь ждёт 5 секунд на ответ!

Решение — использовать асинхронные задачи (Celery, RQ):

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        send_welcome_email_task.delay(instance.id)  # Очередь

@shared_task
def send_welcome_email_task(user_id):
    user = User.objects.get(id=user_id)
    send_email(user.email)

2. Отсутствие гарантии порядка выполнения

Вы не знаете, в каком порядке будут вызваны обработчики, если их несколько:

# обработчик 1
@receiver(post_save, sender=User)
def log_user_creation(sender, instance, created, **kwargs):
    if created:
        print(f"User created: {instance.id}")

# обработчик 2
@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        print(f"Sending email to {instance.email}")

# Порядок вывода не определён!
# Может быть: User created -> Email sent
# Может быть: Email sent -> User created

Решение — явный контроль порядка через dispatch_uid и вызов обработчиков вручную:

# Вместо сигналов
class UserService:
    @staticmethod
    def create_user(email, name):
        user = User.objects.create(email=email, name=name)
        UserService._log_creation(user)
        UserService._send_welcome_email(user)
        return user
    
    @staticmethod
    def _log_creation(user):
        print(f"User created: {user.id}")
    
    @staticmethod
    def _send_welcome_email(user):
        send_email(user.email)

3. Сигналы не отправляются при bulk операциях

# ❌ Сигналы НЕ отправляются
User.objects.filter(is_active=False).delete()
User.objects.bulk_create([user1, user2, user3])
User.objects.bulk_update([user1, user2], fields=['email'])
User.objects.filter(age__lt=18).update(age=18)

# ✅ Сигналы отправляются
for user in User.objects.filter(is_active=False):
    user.delete()  # отправит pre_delete и post_delete

Это может привести к багам, когда ваше кеширование не обновляется после bulk update:

@receiver(post_save, sender=User)
def invalidate_cache(sender, instance, **kwargs):
    cache.delete(f'user_{instance.id}')

# Кеш не инвалидируется!
User.objects.filter(age__lt=18).update(age=18)

Решение:

def update_users_bulk(user_ids, age):
    # Способ 1: Явно инвалидировать кеш
    User.objects.filter(id__in=user_ids).update(age=age)
    for user_id in user_ids:
        cache.delete(f'user_{user_id}')
    
    # Способ 2: Использовать refresh_from_db()
    users = User.objects.filter(id__in=user_ids)
    for user in users:
        user.age = age
        user.save()  # сигнал отправится

4. Исключения в обработчиках прерывают сохранение

@receiver(post_save, sender=User)
def risky_handler(sender, instance, **kwargs):
    result = 1 / 0  # Exception!

# Хотя save() закончилась, обработчик упал
try:
    user = User.objects.create(email='test@example.com')
except ZeroDivisionError:
    print("Handler crashed, but user was created!")
    # Но ты не узнаешь об этом из-за исключения

Решение — обрабатывать исключения в обработчиках:

@receiver(post_save, sender=User)
def safe_handler(sender, instance, **kwargs):
    try:
        some_operation()
    except Exception as e:
        logger.error(f"Signal handler failed: {e}", exc_info=True)
        # Не пробрасываем исключение!

5. Проблемы с тестированием

# Сложно мокировать
from unittest.mock import patch

with patch('myapp.signals.send_email') as mock_email:
    user = User.objects.create(email='test@example.com')
    # Надо понимать, что send_email вызовется в сигнале
    # Это делает тесты хрупкими

Решение — отключать сигналы в тестах:

from django.test import TestCase
from django.db.models.signals import post_save
from myapp.models import User
from myapp.signals import send_welcome_email

class UserTestCase(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        post_save.disconnect(send_welcome_email, sender=User)
    
    def test_user_creation(self):
        user = User.objects.create(email='test@example.com')
        self.assertEqual(user.email, 'test@example.com')

6. Несохранённые изменения в сигналах

@receiver(post_save, sender=User)
def update_related(sender, instance, **kwargs):
    instance.email = 'modified@example.com'  # Только в памяти
    # Это изменение не сохранится в БД!

user = User.objects.create(email='test@example.com')
print(user.email)  # 'modified@example.com' (но в БД — 'test@example.com')

Решение:

@receiver(post_save, sender=User)
def update_related(sender, instance, **kwargs):
    instance.email = 'modified@example.com'
    instance.save()  # Явное сохранение

Рекомендации

1. Используй сигналы только для простых операций (логирование, инвалидация кеша)
2. Для сложной логики лучше использовать сервисные слои (DDD pattern)
3. Избегай рекурсии — добавь dispatch_uid для предотвращения двойной обработки
4. Тестируй с отключенными сигналами — так проще найти баги
5. Для тяжёлых операций используй Celery/RQ
6. Логируй ошибки в обработчиках — не пробрасывай исключения

Сигналы Django — это инструмент для связывания компонентов, но они имеют множество подводных камней. В большинстве случаев явный код (сервисные функции) безопаснее и понятнее.