Señales en Django

Sistema de señales

Las señales de Django implementan el patrón Observer: permiten que componentes desacoplados reciban notificaciones cuando ocurren determinados eventos. Las señales más importantes son las de modelos:

• pre_save: antes de guardar (crear o actualizar)
• post_save: después de guardar
• pre_delete: antes de eliminar
• post_delete: después de eliminar
• m2m_changed: cuando cambia una relación ManyToMany

Conectar señales con @receiver

El decorador @receiver es la forma más limpia de conectar señales:

# signals.py
from django.db.models.signals import post_save, pre_save, post_delete
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from .models import Perfil, Pedido, Producto, HistorialPrecios

Usuario = get_user_model()

@receiver(post_save, sender=Usuario)
def crear_perfil_usuario(sender, instance, created, **kwargs):
    """Crea automáticamente el perfil cuando se crea un usuario."""
    if created:
        Perfil.objects.create(usuario=instance)

@receiver(post_save, sender=Usuario)
def guardar_perfil_usuario(sender, instance, **kwargs):
    """Sincroniza el perfil cuando se guarda el usuario."""
    if hasattr(instance, 'perfil'):
        instance.perfil.save()

pre_save: modificar datos antes de guardar

from django.utils.text import slugify

@receiver(pre_save, sender=Producto)
def generar_slug_producto(sender, instance, **kwargs):
    """Genera el slug automáticamente si no está establecido."""
    if not instance.slug:
        slug_base = slugify(instance.nombre)
        slug = slug_base
        contador = 1
        while Producto.objects.filter(slug=slug).exclude(pk=instance.pk).exists():
            slug = f'{slug_base}-{contador}'
            contador += 1
        instance.slug = slug

@receiver(pre_save, sender=Producto)
def registrar_cambio_precio(sender, instance, **kwargs):
    """Guarda el historial cuando cambia el precio."""
    if instance.pk:  # Solo en actualizaciones (no en creación)
        try:
            precio_anterior = Producto.objects.get(pk=instance.pk).precio
            if precio_anterior != instance.precio:
                HistorialPrecios.objects.create(
                    producto=instance,
                    precio_anterior=precio_anterior,
                    precio_nuevo=instance.precio
                )
        except Producto.DoesNotExist:
            pass

post_save con el flag created

El parámetro created indica si el objeto acaba de ser creado (True) o actualizado (False):

@receiver(post_save, sender=Pedido)
def notificar_pedido(sender, instance, created, **kwargs):
    if created:
        # Email de confirmación de nuevo pedido
        enviar_email_confirmacion_pedido.delay(instance.pk)  # Celery
    elif instance.estado == 'enviado':
        # Email de notificación de envío
        enviar_email_envio.delay(instance.pk)
    elif instance.estado == 'entregado':
        # Solicitar valoración
        enviar_solicitud_valoracion.delay(instance.pk)

post_delete: limpiar recursos

import os

@receiver(post_delete, sender=Producto)
def eliminar_imagen_producto(sender, instance, **kwargs):
    """Elimina la imagen del disco al borrar el producto."""
    if instance.imagen:
        if os.path.isfile(instance.imagen.path):
            os.remove(instance.imagen.path)

@receiver(post_delete, sender=Pedido)
def liberar_stock(sender, instance, **kwargs):
    """Libera el stock cuando se elimina un pedido."""
    from django.db.models import F
    for linea in instance.lineas.all():
        linea.producto.stock = F('stock') + linea.cantidad
        linea.producto.save(update_fields=['stock'])

Señales personalizadas

from django.dispatch import Signal, receiver

# Definir señales personalizadas
pedido_procesado = Signal()  # Enviará: pedido, usuario
pago_confirmado = Signal()   # Enviará: pago, importe

# Enviar una señal personalizada
def procesar_pedido(pedido, usuario):
    # ... lógica de procesamiento ...
    pedido.estado = 'procesado'
    pedido.save()

    # Enviar la señal
    pedido_procesado.send(sender=Pedido, pedido=pedido, usuario=usuario)

# Conectar a la señal personalizada
@receiver(pedido_procesado)
def actualizar_estadisticas_usuario(sender, pedido, usuario, **kwargs):
    usuario.perfil.total_pedidos += 1
    usuario.perfil.save(update_fields=['total_pedidos'])

@receiver(pedido_procesado)
def enviar_factura(sender, pedido, **kwargs):
    generar_y_enviar_factura.delay(pedido.pk)

Conectar señales en AppConfig.ready()

# apps.py
from django.apps import AppConfig

class TiendaConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'tienda'

    def ready(self):
        import tienda.signals  # Importar para conectar las señales

Señales en tests

En los tests, las señales se ejecutan normalmente. Para desactivarlas temporalmente:

from django.test import TestCase
from unittest.mock import patch
from django.db.models.signals import post_save

class ProductoTest(TestCase):
    @patch('tienda.signals.registrar_cambio_precio')
    def test_sin_señal(self, mock_signal):
        """Test que desactiva la señal temporalmente."""
        Producto.objects.create(nombre='Test', precio=9.99)
        mock_signal.assert_not_called()

    def test_con_disconnect(self):
        from tienda.signals import notificar_pedido
        post_save.disconnect(notificar_pedido, sender=Pedido)
        try:
            Pedido.objects.create(...)
        finally:
            post_save.connect(notificar_pedido, sender=Pedido)