Observer en sistemas de pagos con Python

Curso de Patrones de Diseño y SOLID en Python

Contenido del curso

Patrones de Diseño y Principios SOLID

1
Patrones de diseño y SOLID en Python
02:07 min

Principios SOLID

Reestructuración del proyecto

13
Reestructuración con módulos de Python y SOLID
13:32 min

Patrones de Diseño

Tomar examen

Cursos Empresas Blog Live Conf Eventos Precios

0:0011:11

Observer en sistemas de pagos con Python

Curso de Patrones de Diseño y SOLID en Python

Clase 24 de 27

Clase 24 / 27

Recursos

Implementa notificaciones de pagos en Python con confianza: aquí verás cómo aplicar el patrón Observer para informar a múltiples sistemas cuando un pago es exitoso o fallido, usando listeners, manager, Protocol, genéricos y el patrón builder para un diseño limpio y extensible.

¿Qué resuelve el patrón observer en el sistema de pagos?

El objetivo es notificar a distintos sistemas ante dos eventos: pago exitoso y pago fallido. Para lograrlo, se define una interfaz de listener con un método único, notify, y un manager que centraliza la suscripción, desuscripción y la propagación del evento a todos los listeners interesados.

Interfaz mínima y clara: un solo método notify que recibe un evento genérico de tipo T.
Gestión centralizada: un Listeners Manager con métodos subscribe, unsubscribe y notify_all.
Extensibilidad: agregar nuevos listeners es simple y no rompe el código existente.
Integración con el servicio: el builder configura el manager y sus listeners al construir la clase principal.

¿Cómo se implementa el patrón observer paso a paso?

A continuación, una guía práctica basada en el código explicado.

¿Cómo se define la interfaz listener con Protocol y genéricos?

Se crea una carpeta listeners/ y un archivo listener.py que declara el Protocol con un genérico T y el método notify.

# listeners/listener.py
from typing import Protocol, TypeVar

T = TypeVar("T")

class Listener(Protocol[T]):
    def notify(self, event: T) -> None:
        ...

Puntos clave:

Uso de typing.Protocol para definir la forma del listener.
Genérico T para aceptar cualquier tipo de evento sin perder tipado.
El método requiere self y no retorna valor.

¿Cómo gestionar suscripciones con un manager usando dataclass?

Se crea manager.py con un atributo lista de listeners y tres métodos: subscribe, unsubscribe y notify_all. La lista se inicializa vacía con field(default_factory=list).

# listeners/manager.py
from dataclasses import dataclass, field
from typing import List, TypeVar
from .listener import Listener

T = TypeVar("T")

@dataclass
class ListenersManager:
    listeners: List[Listener[T]] = field(default_factory=list)

    def subscribe(self, listener: Listener[T]) -> None:
        self.listeners.append(listener)

    def unsubscribe(self, listener: Listener[T]) -> None:
        self.listeners.remove(listener)

    def notify_all(self, event: T) -> None:
        for listener in self.listeners:
            listener.notify(event)

Puntos clave:

dataclass simplifica la construcción de la clase.
default_factory=list evita pasar la lista por parámetro y previene listas mutables compartidas.
notify_all itera y llama a notify en cada listener suscrito.

¿Cómo crear un listener concreto para contabilidad?

Se implementa un listener sencillo, por ejemplo, AccountabilityListener, que imprime el evento recibido.

# listeners/accountability_listener.py
from .listener import Listener

class AccountabilityListener(Listener[object]):
    def notify(self, event: object) -> None:
        print("Notificando el evento:", event)

Puntos clave:

Implementa notify con la lógica específica (aquí, un print controlado).
El tipo de evento puede ajustarse según el dominio.

¿Cómo integrarlo en el servicio con el patrón builder?

Se agrega al service un atributo de tipo ListenersManager y un método del builder para configurarlo y suscribir listeners por defecto.

# service_builder.py (fragmento)
from typing import Optional
from listeners.manager import ListenersManager
from listeners.accountability_listener import AccountabilityListener

class ServiceBuilder:
    def __init__(self) -> None:
        self.listeners: Optional[ListenersManager] = None

    def set_listeners(self) -> "ServiceBuilder":
        manager = ListenersManager()
        accountability = AccountabilityListener()
        manager.subscribe(accountability)
        self.listeners = manager
        return self

En el flujo de pagos, se notifica en una transacción exitosa después de obtener el payment_response dentro de process_transaction:

# dentro del servicio, luego de un pago exitoso
self.listeners.notify_all(f"pago exitoso: {payment_response.transaction_id}")

Sugerencia pendiente: notificar también el pago fallido cuando el payment_response lo indique.

¿Qué habilidades y keywords se refuerzan en esta implementación?

La construcción refuerza conceptos de diseño y tipado en Python que facilitan el mantenimiento y la escalabilidad.

Patrón Observer: desacopla emisores de eventos y suscriptores.
Listener como Protocol: interfaz explícita con un único método notify.
Genéricos (T): eventos de cualquier tipo con tipado estático.
Listeners Manager: métodos subscribe, unsubscribe, notify_all para orquestación.
dataclass y field(default_factory=list): inicialización segura de colecciones.
AccountabilityListener: ejemplo de implementación concreta.
Módulo listeners con init.py: facilita importaciones limpias.
Builder: configuración progresiva del service para agregar dependencias como listeners.
Eventos de dominio: "pago exitoso" y "pago fallido" con datos como transaction_id.

¿Ya resolviste la notificación de pago fallido? Cuéntame en los comentarios cómo lo implementaste y qué diseño elegiste para el evento.

Mario Alexander Vargas Celis

Hace 2 años

Implementar el **Patrón Observer** implica crear un sistema donde un objeto central (*subject*) notifica automáticamente a varios objetos suscritos (*observers*) cada vez que su estado cambia. Veamos cómo implementarlo paso a paso en Python. ### Escenario: Sistema de Notificaciones para Pedidos Vamos a construir un ejemplo donde un sistema de pedidos notifica a varios servicios (correo electrónico, SMS, registro en el log) cuando se procesa un pedido. ### Paso 1: Crear el Subject El *subject* gestiona la lista de *observers* y los notifica cuando ocurre un cambio.

    def attach(self, observer):        """Suscribe un observer al subject."""        self.\_observers.append(observer)
    def detach(self, observer):        """Elimina un observer del subject."""        self.\_observers.remove(observer)
    def notify(self, message):        """Notifica a todos los observers."""        for observer in self.\_observers:            observer.update(message)```
\### Paso 2: Crear la Interfaz del Observer
Define una clase base para los *\*observers\**. Esto asegura que todos implementen el método `update`.
```pythonfrom abc import ABC, abstractmethod
class Observer(ABC):    @abstractmethod    def update(self, message):        """Define cómo reaccionará el observer al mensaje."""        pass```
\### Paso 3: Crear Observers Concretos
Cada *\*observer\** define su propia implementación del método `update`.
```pythonclass EmailNotifier(Observer):    def update(self, message):        print(f"EmailNotifier: Enviando email con el mensaje: '{message}'")
class SMSNotifier(Observer):    def update(self, message):        print(f"SMSNotifier: Enviando SMS con el mensaje: '{message}'")
class LogNotifier(Observer):    def update(self, message):        print(f"LogNotifier: Registrando en el log: '{message}'")```
\### Paso 4: Crear un Subject Concreto
Crea una clase que represente el *\*subject\**. En este caso, un sistema de pedidos.
```pythonclass OrderSystem(Subject):    def process\_order(self, order\_id):        print(f"Procesando pedido {order\_id}...")        self.notify(f"Pedido {order\_id} ha sido procesado con éxito.")```
\### Paso 5: Integrar Todo
Conecta los *\*observers\** al *\*subject\** y prueba el sistema.
```python# Crear el sistema de pedidosorder\_system = OrderSystem()
\# Crear los observersemail\_notifier = EmailNotifier()sms\_notifier = SMSNotifier()log\_notifier = LogNotifier()
\# Suscribir los observers al sistema de pedidosorder\_system.attach(email\_notifier)order\_system.attach(sms\_notifier)order\_system.attach(log\_notifier)
\# Procesar un pedidoorder\_system.process\_order("12345")
\# Eliminar un observerorder\_system.detach(sms\_notifier)
\# Procesar otro pedidoorder\_system.process\_order("67890")```
\### Salida Esperada
\*\*Primer pedido (todos los observers suscritos):\*\*
```Procesando pedido 12345...EmailNotifier: Enviando email con el mensaje: 'Pedido 12345 ha sido procesado con éxito.'SMSNotifier: Enviando SMS con el mensaje: 'Pedido 12345 ha sido procesado con éxito.'LogNotifier: Registrando en el log: 'Pedido 12345 ha sido procesado con éxito.'```
\*\*Segundo pedido (SMSNotifier eliminado):\*\*
```Procesando pedido 67890...EmailNotifier: Enviando email con el mensaje: 'Pedido 67890 ha sido procesado con éxito.'LogNotifier: Registrando en el log: 'Pedido 67890 ha sido procesado con éxito.'```
\### Ventajas del Patrón Observer
1\. \*\*Desacoplamiento\*\*: Los *\*subjects\** no necesitan saber cómo funcionan los *\*observers\**.2. \*\*Flexibilidad\*\*: Puedes añadir o eliminar *\*observers\** dinámicamente.3. \*\*Reutilización\*\*: Los *\*observers\** son reutilizables en diferentes sistemas.
\### Casos de Uso Reales
\- \*\*Interfaces gráficas de usuario\*\*: Actualización de vistas cuando el modelo cambia.- \*\*Sistemas de eventos\*\*: Publicación/suscripción (*\*pub-sub\**), como en servicios de mensajería.- \*\*Notificaciones en tiempo real\*\*: Alertas en aplicaciones.
\### Conclusión
El \*\*Patrón Observer\*\* es ideal para sistemas en los que un cambio en un objeto afecta a otros. Su implementación en Python es directa y altamente extensible, lo que lo hace una herramienta poderosa para sistemas desacoplados y dinámicos.

Patrones de Diseño y Principios SOLID

Patrones de diseño y SOLID en Python

Principios SOLID

Qué es el principio de responsabilidad única

Procesador de pagos en Python con Stripe

Cómo aplicar SRP en un procesador de pagos con Stripe

Principio open-closed en Python

Clases abstractas y principio abierto-cerrado

Principio de Liskov en S.O.L.I.D.

Principio de sustitución de Liskov en Python

Principio de segregación de interfaces en SOLID

Segregación de interfaces en procesadores de pagos

Principio de inversión de dependencias en SOLID

Principio de inversión de dependencias: servicio de pagos flexible

Reestructuración del proyecto

Reestructuración con módulos de Python y SOLID

Patrones de Diseño

Qué son los patrones de diseño: definición y categorías

Strategy Pattern con Python y setprocessor

Strategy Pattern para pagos en Python

Factory Pattern: centralizar creación de objetos

Patrón Factory para procesar pagos con match

Qué es el patrón Decorator

Patrón decorador en servicios de pagos

Builder Pattern: construcción paso a paso

Builder Pattern aplicado a un servicio de pagos

Observer Pattern en sistemas de eventos

Observer en sistemas de pagos con Python

Chain of Responsibility para validar pagos

Chain of Responsibility en servicios de pagos

Código funcional vs código mantenible en Python

Observer en sistemas de pagos con Python

Progreso del curso

Patrones de Diseño y Principios SOLID

Principios SOLID

Reestructuración del proyecto

Patrones de Diseño

Recursos

Recursos