Observer en sistemas de pagos con Python

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.