🚀 Mini-Tutorial: Sistema de Notificaciones Inteligente con SOLID + Patrones de Diseño

🎯 ¿Qué vamos a construir?

Un sistema flexible de notificaciones que puede enviar mensajes por diferentes canales (Email, SMS, Push, Slack) aplicando todos los principios SOLID y combinando 3 patrones de diseño en una sola solución elegante.

Lo que hace especial este tutorial:

• Combina Strategy + Chain of Responsibility + Observer
• Aplica los 5 principios SOLID simultáneamente
• Caso de uso real y práctico
• Código production-ready

📚 Prerequisitos

Debes conocer:

• ✅ Los 5 principios SOLID
• ✅ Strategy Pattern
• ✅ Observer Pattern
• ✅ Chain of Responsibility Pattern

🏗️ Arquitectura del Sistema

Usuario → NotificationService → [Chain] → [Strategy] → Canal específico
                                    ↓
                                [Observer] → Analytics

💡 El Problema

Imagina que tienes una app y necesitas:

1. Enviar notificaciones por múltiples canales
2. Aplicar reglas de validación antes de enviar
3. Registrar analytics de cada notificación
4. Poder agregar nuevos canales sin modificar código existente
5. Poder cambiar el canal en runtime

🔨 Paso 1: Abstracciones (SRP + OCP + DIP)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List
from enum import Enum


class NotificationPriority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"


@dataclass
class NotificationMessage:
    """Modelo de datos para notificaciones"""
    recipient: str
    subject: str
    body: str
    priority: NotificationPriority
    metadata: dict = None


# Principio de Inversión de Dependencias (DIP)
class NotificationChannel(ABC):
    """Abstracción base para todos los canales"""
    
    @abstractmethod
    def send(self, message: NotificationMessage) -> bool:
        """Envía una notificación por el canal específico"""
        pass
    
    @abstractmethod
    def get_channel_name(self) -> str:
        """Retorna el nombre del canal"""
        pass

¿Por qué esto es bueno?

• SRP: Cada clase tiene una única responsabilidad
• OCP: Podemos extender con nuevos canales sin modificar la interfaz
• DIP: Dependemos de abstracciones, no de implementaciones concretas

🔨 Paso 2: Implementaciones Concretas (LSP)

class EmailChannel(NotificationChannel):
    """Canal de notificaciones por Email"""
    
    def send(self, message: NotificationMessage) -> bool:
        print(f"📧 Enviando email a {message.recipient}")
        print(f"   Asunto: {message.subject}")
        print(f"   Prioridad: {message.priority.value}")
        # Aquí iría la lógica real de envío con SMTP o servicio externo
        return True
    
    def get_channel_name(self) -> str:
        return "Email"


class SMSChannel(NotificationChannel):
    """Canal de notificaciones por SMS"""
    
    def send(self, message: NotificationMessage) -> bool:
        print(f"📱 Enviando SMS a {message.recipient}")
        print(f"   Mensaje: {message.body[:50]}...")
        # Aquí iría la lógica real con Twilio, SNS, etc.
        return True
    
    def get_channel_name(self) -> str:
        return "SMS"


class PushChannel(NotificationChannel):
    """Canal de notificaciones Push"""
    
    def send(self, message: NotificationMessage) -> bool:
        print(f"🔔 Enviando Push a {message.recipient}")
        print(f"   {message.subject}: {message.body}")
        # Aquí iría FCM, OneSignal, etc.
        return True
    
    def get_channel_name(self) -> str:
        return "Push"


class SlackChannel(NotificationChannel):
    """Canal de notificaciones por Slack"""
    
    def send(self, message: NotificationMessage) -> bool:
        print(f"💬 Enviando mensaje a Slack #{message.recipient}")
        print(f"   {message.subject}\n   {message.body}")
        # Aquí iría Slack Webhook
        return True
    
    def get_channel_name(self) -> str:
        return "Slack"

Principio de Sustitución de Liskov (LSP):
Cualquier canal puede reemplazar a otro sin romper el sistema. Todos cumplen el contrato de NotificationChannel.

🔨 Paso 3: Chain of Responsibility para Validaciones

class ValidationHandler(ABC):
    """Handler base para la cadena de validación"""
    
    def __init__(self):
        self._next_handler: Optional[ValidationHandler] = None
    
    def set_next(self, handler: 'ValidationHandler') -> 'ValidationHandler':
        self._next_handler = handler
        return handler
    
    @abstractmethod
    def validate(self, message: NotificationMessage) -> tuple[bool, str]:
        """Valida el mensaje. Retorna (es_válido, mensaje_error)"""
        pass
    
    def handle(self, message: NotificationMessage) -> tuple[bool, str]:
        is_valid, error_msg = self.validate(message)
        
        if not is_valid:
            return False, error_msg
        
        if self._next_handler:
            return self._next_handler.handle(message)
        
        return True, "Todas las validaciones pasaron"


class RecipientValidator(ValidationHandler):
    """Valida que el destinatario no esté vacío"""
    
    def validate(self, message: NotificationMessage) -> tuple[bool, str]:
        if not message.recipient or len(message.recipient.strip()) == 0:
            return False, "El destinatario no puede estar vacío"
        return True, ""


class ContentValidator(ValidationHandler):
    """Valida que haya contenido en el mensaje"""
    
    def validate(self, message: NotificationMessage) -> tuple[bool, str]:
        if not message.body or len(message.body.strip()) == 0:
            return False, "El mensaje no puede estar vacío"
        return True, ""


class PriorityValidator(ValidationHandler):
    """Valida mensajes urgentes (ejemplo: solo en horario laboral)"""
    
    def validate(self, message: NotificationMessage) -> tuple[bool, str]:
        from datetime import datetime
        
        if message.priority == NotificationPriority.URGENT:
            current_hour = datetime.now().hour
            if current_hour < 8 or current_hour > 20:
                return False, "Notificaciones urgentes solo entre 8am y 8pm"
        
        return True, ""

Chain of Responsibility en acción:
Cada validador solo se preocupa de UNA cosa (SRP) y puede pasarle el control al siguiente. Fácil agregar nuevas validaciones sin modificar las existentes (OCP).

🔨 Paso 4: Observer Pattern para Analytics

class NotificationObserver(ABC):
    """Observador base para eventos de notificación"""
    
    @abstractmethod
    def on_notification_sent(
        self, 
        channel: str, 
        message: NotificationMessage,
        success: bool
    ) -> None:
        pass


class AnalyticsObserver(NotificationObserver):
    """Registra métricas de notificaciones"""
    
    def __init__(self):
        self.sent_count = 0
        self.failed_count = 0
        self.channels_used = {}
    
    def on_notification_sent(
        self, 
        channel: str, 
        message: NotificationMessage,
        success: bool
    ) -> None:
        if success:
            self.sent_count += 1
            self.channels_used[channel] = self.channels_used.get(channel, 0) + 1
        else:
            self.failed_count += 1
        
        print(f"📊 [Analytics] Total enviadas: {self.sent_count}, "
              f"Fallidas: {self.failed_count}")
    
    def get_report(self) -> dict:
        return {
            "total_sent": self.sent_count,
            "total_failed": self.failed_count,
            "channels": self.channels_used
        }


class LoggerObserver(NotificationObserver):
    """Registra logs de todas las notificaciones"""
    
    def on_notification_sent(
        self, 
        channel: str, 
        message: NotificationMessage,
        success: bool
    ) -> None:
        status = "✅ ENVIADA" if success else "❌ FALLIDA"
        print(f"📝 [Log] {status} - Canal: {channel}, "
              f"Para: {message.recipient}, "
              f"Prioridad: {message.priority.value}")

🔨 Paso 5: Servicio Principal (Strategy + ISP)

class NotificationService:
    """
    Servicio principal que orquesta todo el sistema
    Aplica Strategy Pattern para cambiar canales dinámicamente
    """
    
    def __init__(self, default_channel: NotificationChannel):
        self._channel = default_channel  # Strategy Pattern
        self._validators: Optional[ValidationHandler] = None
        self._observers: List[NotificationObserver] = []
    
    # Strategy Pattern: cambiar canal en runtime
    def set_channel(self, channel: NotificationChannel) -> None:
        """Cambia el canal de notificación (Strategy Pattern)"""
        print(f"🔄 Cambiando canal a: {channel.get_channel_name()}")
        self._channel = channel
    
    # Chain of Responsibility: configurar validaciones
    def set_validation_chain(self, first_validator: ValidationHandler) -> None:
        """Configura la cadena de validación"""
        self._validators = first_validator
    
    # Observer Pattern: suscribir observadores
    def add_observer(self, observer: NotificationObserver) -> None:
        """Agrega un observador de eventos"""
        self._observers.append(observer)
    
    def send_notification(self, message: NotificationMessage) -> bool:
        """
        Envía una notificación aplicando:
        1. Validaciones (Chain of Responsibility)
        2. Envío por canal (Strategy)
        3. Notificación a observadores (Observer)
        """
        
        # 1. Validar con Chain of Responsibility
        if self._validators:
            is_valid, error_msg = self._validators.handle(message)
            if not is_valid:
                print(f"❌ Validación fallida: {error_msg}")
                self._notify_observers(False, message)
                return False
        
        # 2. Enviar usando Strategy
        try:
            success = self._channel.send(message)
            
            # 3. Notificar a observers
            self._notify_observers(success, message)
            
            return success
            
        except Exception as e:
            print(f"❌ Error al enviar: {str(e)}")
            self._notify_observers(False, message)
            return False
    
    def _notify_observers(self, success: bool, message: NotificationMessage) -> None:
        """Notifica a todos los observadores"""
        channel_name = self._channel.get_channel_name()
        for observer in self._observers:
            observer.on_notification_sent(channel_name, message, success)

🚀 Paso 6: ¡Úsalo en producción!

def main():
    """Demo del sistema completo"""
    
    print("="*60)
    print("🎯 SISTEMA DE NOTIFICACIONES INTELIGENTE")
    print("="*60)
    
    # 1. Configurar el servicio con canal por defecto
    service = NotificationService(default_channel=EmailChannel())
    
    # 2. Configurar Chain of Responsibility para validaciones
    recipient_validator = RecipientValidator()
    content_validator = ContentValidator()
    priority_validator = PriorityValidator()
    
    recipient_validator.set_next(content_validator).set_next(priority_validator)
    service.set_validation_chain(recipient_validator)
    
    # 3. Agregar Observers para analytics y logs
    analytics = AnalyticsObserver()
    logger = LoggerObserver()
    
    service.add_observer(analytics)
    service.add_observer(logger)
    
    print("\n--- Test 1: Email con validación exitosa ---")
    message1 = NotificationMessage(
        recipient="usuario@example.com",
        subject="Bienvenido a Platzi",
        body="Gracias por registrarte en nuestra plataforma",
        priority=NotificationPriority.MEDIUM
    )
    service.send_notification(message1)
    
    print("\n--- Test 2: Cambiar a SMS (Strategy Pattern) ---")
    service.set_channel(SMSChannel())
    message2 = NotificationMessage(
        recipient="+573001234567",
        subject="Código de verificación",
        body="Tu código es: 123456",
        priority=NotificationPriority.HIGH
    )
    service.send_notification(message2)
    
    print("\n--- Test 3: Validación fallida (mensaje vacío) ---")
    message3 = NotificationMessage(
        recipient="test@test.com",
        subject="Test",
        body="",  # Esto fallará
        priority=NotificationPriority.LOW
    )
    service.send_notification(message3)
    
    print("\n--- Test 4: Cambiar a Slack ---")
    service.set_channel(SlackChannel())
    message4 = NotificationMessage(
        recipient="general",
        subject="🚀 Deployment exitoso",
        body="La versión 2.0 está en producción",
        priority=NotificationPriority.MEDIUM
    )
    service.send_notification(message4)
    
    print("\n--- Test 5: Múltiples canales ---")
    channels = [PushChannel(), EmailChannel(), SMSChannel()]
    message5 = NotificationMessage(
        recipient="admin@platzi.com",
        subject="Alerta de Sistema",
        body="Uso de CPU al 90%",
        priority=NotificationPriority.URGENT
    )
    
    for channel in channels:
        service.set_channel(channel)
        service.send_notification(message5)
    
    # Mostrar reporte de analytics
    print("\n" + "="*60)
    print("📊 REPORTE DE ANALYTICS")
    print("="*60)
    report = analytics.get_report()
    print(f"Total enviadas: {report['total_sent']}")
    print(f"Total fallidas: {report['total_failed']}")
    print(f"Canales usados: {report['channels']}")
    print("="*60)


if __name__ == "__main__":
    main()

🎯 ¿Qué acabas de aprender?

✅ Principios SOLID aplicados:

1. SRP: Cada clase tiene UNA responsabilidad

   • EmailChannel solo maneja emails
   • RecipientValidator solo valida destinatarios
   • AnalyticsObserver solo registra métricas

2. OCP: Abierto a extensión, cerrado a modificación

   • Puedes agregar WhatsAppChannel sin tocar código existente
   • Puedes agregar RateLimitValidator sin modificar otras validaciones

3. LSP: Cualquier NotificationChannel puede sustituir a otro

   • El sistema funciona igual con Email, SMS, Push o Slack

4. ISP: Interfaces segregadas

   • NotificationChannel solo tiene lo necesario para enviar
   • No obliga a implementar métodos innecesarios

5. DIP: Dependemos de abstracciones

   • NotificationService depende de NotificationChannel (abstracción)
   • No depende de EmailChannel o SMSChannel (concretas)

🎨 Patrones de Diseño usados:

1. Strategy Pattern:

   • Cambiar canal de notificación en runtime
   • service.set_channel(SMSChannel())

2. Chain of Responsibility:

   • Validaciones secuenciales y desacopladas
   • Cada validador decide si continúa o detiene la cadena

3. Observer Pattern:

   • Analytics y logs sin acoplar al servicio principal
   • Fácil agregar nuevos observers (ej: métricas a DataDog)

🚀 Siguiente Nivel: Extensiones

1. Factory Pattern para crear canales

class NotificationChannelFactory:
    @staticmethod
    def create_channel(channel_type: str) -> NotificationChannel:
        channels = {
            "email": EmailChannel,
            "sms": SMSChannel,
            "push": PushChannel,
            "slack": SlackChannel
        }
        return channels[channel_type.lower()]()

# Uso
service.set_channel(NotificationChannelFactory.create_channel("email"))

2. Decorator Pattern para retry logic

class RetryDecorator(NotificationChannel):
    def __init__(self, channel: NotificationChannel, retries: int = 3):
        self._channel = channel
        self._retries = retries
    
    def send(self, message: NotificationMessage) -> bool:
        for attempt in range(self._retries):
            try:
                return self._channel.send(message)
            except Exception:
                if attempt == self._retries - 1:
                    raise
        return False
    
    def get_channel_name(self) -> str:
        return f"Retry({self._channel.get_channel_name()})"

# Uso
reliable_email = RetryDecorator(EmailChannel(), retries=3)
service.set_channel(reliable_email)

3. Template Method para canales con formato común

class FormattedChannel(NotificationChannel):
    def send(self, message: NotificationMessage) -> bool:
        formatted = self.format_message(message)
        return self.send_formatted(formatted)
    
    @abstractmethod
    def format_message(self, message: NotificationMessage) -> str:
        pass
    
    @abstractmethod
    def send_formatted(self, formatted_message: str) -> bool:
        pass

💡 Casos de Uso Reales

1. E-commerce: Notificar pedidos por Email + SMS + Push
2. SaaS: Alertas de sistema por Slack + Email según prioridad
3. Banking: OTPs por SMS con validación estricta
4. Healthcare: Recordatorios de citas multi-canal
5. DevOps: Alertas de monitoreo por múltiples canales

🎓 Conclusión

Has construido un sistema production-ready que:

• ✅ Es extensible (agregar canales sin dolor)
• ✅ Es mantenible (código limpio y organizado)
• ✅ Es testeable (interfaces claras)
• ✅ Es flexible (cambiar comportamiento en runtime)
• ✅ Sigue SOLID y patrones de diseño

Pro Tip: Este mismo patrón se puede aplicar a:

• Sistemas de pagos (como en el curso)
• Exportadores de datos (PDF, Excel, CSV)
• Generadores de reportes
• Sistemas de autenticación (OAuth, JWT, API Key)

📚 Referencias

• Curso de Patrones de Diseño y SOLID en Python - Platzi
• Refactoring Guru - Design Patterns
• Python Type Hints - PEP 484

¿Te gustó este tutorial?

• 🌟 Compártelo con la comunidad
• 💬 Deja tus comentarios y mejoras
• 🚀 Implementa tu propia versión