Segregación de interfaces en procesadores de pagos

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:009:05

Segregación de interfaces en procesadores de pagos

Curso de Patrones de Diseño y SOLID en Python

Clase 10 de 27

Clase 10 / 27

Comentarios (12)12 comentarios

Juan Camilo Suarez

Hace 2 años

Hola :D Recreé el archivo before.py para que puedas usarlo:

"""
features agregados: 
+ PaymentResponse: objeto
+ RefundPaymentProtocol: interfaz
+ PaymentProcessorProtocol:interfaz
+ RecurringPaymentProtocol: interfaz
+ StripePaymentProcessor: subclase (ejecuta demasiadas cosas)
+ OfflinePaymentProcessor: subclase
-------------------------------------------------------------
PREGUNTA: ¿Qué principio SOLID se estaba siguiendo en los métodos de recurrencia?
El principio S: single responsibity principle porque está dedicado úni-
camente a configurar los pagos con stripe y de la suscripción 
están delegadas a métodos auxiliares
"""

import os
from dataclasses import dataclass, field
from typing import Optional, Protocol
import uuid
import stripe
from dotenv import load_dotenv
from pydantic import BaseModel
from stripe.error import StripeError

_ = load_dotenv()


class ContactInfo(BaseModel):
    email: Optional[str] = None
    phone: Optional[str] = None


class CustomerData(BaseModel):
    name: str
    contact_info: ContactInfo
    customer_id: Optional[str] = None


class PaymentData(BaseModel):
    amount: int
    source: str


class PaymentResponse(BaseModel):
    status: str
    amount: int
    transaction_id: Optional[str] = None
    message: Optional[str] = None


class PaymentProcessor(Protocol): 
    """
    Protocol for processing payments, refunds, and recurring payments.
    # HEREDA A TODOS LOS PAYMENTPROCESOR
    This protocol defines the interface for payment processors. Implementations
    should provide methods for processing payments, refunds, and setting up recurring payments.
    """

    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse: ...
    
    def refund_payment(self, transaction_id:str) -> PaymentResponse: ...
    
    def setup_recurring_payment(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) ->PaymentResponse: ...


class StripePaymentProcessor(PaymentProcessor):
    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        try:
            charge = stripe.Charge.create(
                amount=payment_data.amount,
                currency="usd",
                source=payment_data.source,
                description="Charge for " + customer_data.name,
            )
            print("Payment successful")
            return PaymentResponse(
                status=charge["status"],
                amount=charge["amount"],
                transaction_id=charge["id"],
                message="Payment successful",
            )
        except StripeError as e:
            print("Payment failed:", e)
            return PaymentResponse(
                status="failed",
                amount=payment_data.amount,
                transaction_id=None,
                message=str(e),
            )

    def refund_payment(self, transaction_id: str) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        try:
            refund = stripe.Refund.create(charge=transaction_id)
            print("Refund successful")
            return PaymentResponse(
                status=refund["status"],
                amount=refund["amount"],
                transaction_id=refund["id"],
                message="Refund successful",
            )
        except StripeError as e:
            print("Refund failed:", e)
            return PaymentResponse(
                status="failed",
                amount=0,
                transaction_id=None,
                message=str(e),
            )

    def setup_recurring_payment(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        price_id = os.getenv("STRIPE_PRICE_ID", "")
        try:
            customer = self._get_or_create_customer(customer_data)

            payment_method = self._attach_payment_method(
                customer.id, payment_data.source
            )

            self._set_default_payment_method(customer.id, payment_method.id)

            subscription = stripe.Subscription.create(
                customer=customer.id,
                items=[
                    {"price": price_id},
                ],
                expand=["latest_invoice.payment_intent"],
            )

            print("Recurring payment setup successful")
            amount = subscription["items"]["data"][0]["price"]["unit_amount"]
            return PaymentResponse(
                status=subscription["status"],
                amount=amount,
                transaction_id=subscription["id"],
                message="Recurring payment setup successful",
            )
        except StripeError as e:
            print("Recurring payment setup failed:", e)
            return PaymentResponse(
                status="failed",
                amount=0,
                transaction_id=None,
                message=str(e),
            )

    def _get_or_create_customer(
        self, customer_data: CustomerData
    ) -> stripe.Customer:
        """
        Creates a new customer in Stripe or retrieves an existing one.
        """
        if customer_data.customer_id:
            customer = stripe.Customer.retrieve(customer_data.customer_id)
            print(f"Customer retrieved: {customer.id}")
        else:
            if not customer_data.contact_info.email:
                raise ValueError("Email required for subscriptions")
            customer = stripe.Customer.create(
                name=customer_data.name, email=customer_data.contact_info.email
            )
            print(f"Customer created: {customer.id}")
        return customer

    def _attach_payment_method(
        self, customer_id: str, payment_source: str
    ) -> stripe.PaymentMethod:
        """
        Attaches a payment method to a customer.
        """
        payment_method = stripe.PaymentMethod.retrieve(payment_source)
        stripe.PaymentMethod.attach(
            payment_method.id,
            customer=customer_id,
        )
        print(
            f"Payment method {payment_method.id} attached to customer {customer_id}"
        )
        return payment_method

    def _set_default_payment_method(
        self, customer_id: str, payment_method_id: str
    ) -> None:
        """
        Sets the default payment method for a customer.
        """
        stripe.Customer.modify(
            customer_id,
            invoice_settings={
                "default_payment_method": payment_method_id,
            },
        )
        print(f"Default payment method set for customer {customer_id}")


class OfflinePaymentProcessor(PaymentProcessor):
    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        print("Processing offline payment for", customer_data.name)
        return PaymentResponse(
            status="success",
            amount=payment_data.amount,
            transaction_id=str(uuid.uuid4()),
            message="Offline payment success",
        )
    """
    los siguientes métodos levantan errores porque no se pueden hacer reembolsos ni recurrencias a efectivo
    se incumple el principio de segregación de interfaces porque una clase no debería depender 
    de clases que no puede implementar
    """
    def refund_payment(self, transaction_id: str) -> PaymentResponse:
        print("refunds aren't supported in OfflinePaymentOricessor.")
        raise NotImplementedError("Refunds not supported in offline processor.")
    
    def setup_recurring_payment(self, customer_data: CustomerData, payment_data: PaymentData
    ) ->PaymentResponse:
        print("recurring payments aren't supported in OfflinePaymentOricessor.")
        raise NotImplementedError("Refunds not supported in offline processor.")
    

class Notifier(Protocol):
    """
    Protocol for sending notifications.

    This protocol defines the interface for notifiers. Implementations
    should provide a method `send_confirmation` that sends a confirmation
    to the customer.
    """

    def send_confirmation(self, customer_data: CustomerData): ...


class EmailNotifier:
    def send_confirmation(self, customer_data: CustomerData):
        from email.mime.text import MIMEText

        if not customer_data.contact_info.email:
            raise ValueError("Email address is requiered to send an email")

        msg = MIMEText("Thank you for your payment.")
        msg["Subject"] = "Payment Confirmation"
        msg["From"] = "no-reply@example.com"
        msg["To"] = customer_data.contact_info.email

        print("Email sent to", customer_data.contact_info.email)


@dataclass
class SMSNotifier:
    gateway: str

    def send_confirmation(self, customer_data: CustomerData):
        phone_number = customer_data.contact_info.phone
        if not phone_number:
            print("No phone number provided")
            return
        print(
            f"SMS sent to {phone_number} via {self.gateway}: Thank you for your payment."
        )


class TransactionLogger:
    def log_transaction(
        self,
        customer_data: CustomerData,
        payment_data: PaymentData,
        payment_response: PaymentResponse,
    ):
        with open("transactions.log", "a") as log_file:
            log_file.write(
                f"{customer_data.name} paid {payment_data.amount}\n"
            )
            log_file.write(f"Payment status: {payment_response.status}\n")
            if payment_response.transaction_id:
                log_file.write(
                    f"Transaction ID: {payment_response.transaction_id}\n"
                )
            log_file.write(f"Message: {payment_response.message}\n")

    def log_refund(
        self, transaction_id: str, refund_response: PaymentResponse
    ):
        with open("transactions.log", "a") as log_file:
            log_file.write(
                f"Refund processed for transaction {transaction_id}\n"
            )
            log_file.write(f"Refund status: {refund_response.status}\n")
            log_file.write(f"Message: {refund_response.message}\n")


class CustomerValidator:
    def validate(self, customer_data: CustomerData):
        if not customer_data.name:
            print("Invalid customer data: missing name")
            raise ValueError("Invalid customer data: missing name")
        if not customer_data.contact_info:
            print("Invalid customer data: missing contact info")
            raise ValueError("Invalid customer data: missing contact info")
        if not (
            customer_data.contact_info.email
            or customer_data.contact_info.phone
        ):
            print("Invalid customer data: missing email and phone")
            raise ValueError("Invalid customer data: missing email and phone")


class PaymentDataValidator:
    def validate(self, payment_data: PaymentData):
        if not payment_data.source:
            print("Invalid payment data: missing source")
            raise ValueError("Invalid payment data: missing source")
        if payment_data.amount <= 0:
            print("Invalid payment data: amount must be positive")
            raise ValueError("Invalid payment data: amount must be positive")


@dataclass
class PaymentService:
    payment_processor: PaymentProcessor
    notifier: Notifier
    customer_validator: CustomerValidator = field(
        default_factory=CustomerValidator
    )
    payment_validator: PaymentDataValidator = field(
        default_factory=PaymentDataValidator
    )
    logger: TransactionLogger = field(default_factory=TransactionLogger)

    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        self.customer_validator.validate(customer_data)
        self.payment_validator.validate(payment_data)
        payment_response = self.payment_processor.process_transaction(
            customer_data, payment_data
        )
        self.notifier.send_confirmation(customer_data)
        self.logger.log_transaction(
            customer_data, payment_data, payment_response
        )
        return payment_response

    def process_refund(self, transaction_id: str):
        refund_response = self.payment_processor.refund_payment(transaction_id)
        self.logger.log_refund(transaction_id, refund_response)
        return refund_response

    def setup_recurring(
        self, customer_data: CustomerData, payment_data: PaymentData
    ):
        recurring_response = self.payment_processor.setup_recurring_payment(
            customer_data, payment_data
        )
        self.logger.log_transaction(
            customer_data, payment_data, recurring_response
        )
        return recurring_response


if __name__ == "__main__":
    # Set up the payment processors
    stripe_processor = StripePaymentProcessor()
    offline_processor = OfflinePaymentProcessor()

    # Set up the customer data and payment data
    customer_data_with_email = CustomerData(
        name="John Doe", contact_info=ContactInfo(email="john@example.com")
    )
    customer_data_with_phone = CustomerData(
        name="Jane Doe", contact_info=ContactInfo(phone="1234567890")
    )

    # Set up the payment data
    payment_data = PaymentData(amount=100, source="tok_visa")

    # Set up the notifiers
    email_notifier = EmailNotifier()

    sms_gateway = "YourSMSService"
    sms_notifier = SMSNotifier(sms_gateway)

    # # Using Stripe processor with email notifier
    payment_service_email = PaymentService(stripe_processor, email_notifier)
    payment_service_email.process_transaction(customer_data_with_email, payment_data)

    # Using Stripe processor with SMS notifier
    payment_service_sms = PaymentService(stripe_processor, sms_notifier)
    sms_payment_response = payment_service_sms.process_transaction(customer_data_with_phone, payment_data)

    #Using strupe processor with SMS notifier
    payment_service_sms = PaymentService(stripe_processor, sms_notifier)
    sms_payment_response = payment_service_sms.process_transaction(customer_data_with_phone, payment_data)


    # Example of processing a refund using Stripe processor
    transaction_id_to_refund = sms_payment_response.transaction_id
    if transaction_id_to_refund:
        payment_service_email.process_refund(transaction_id_to_refund)

    # Using offline processor with email notifier
    offline_payment_service = PaymentService(offline_processor, email_notifier)
    offline_payment_response = offline_payment_service.process_transaction(
        customer_data_with_email, payment_data
    )

    # Attempt to refund using offline processor (will fail)
    try:
        if offline_payment_response.transaction_id:
            offline_payment_service.process_refund(
                offline_payment_response.transaction_id
            )
    except Exception as e:
        print(f"Refund failed and PaymentService raised an exception: {e}")

    # Attempt to set up recurring payment using offline processor (will fail)
    try:
        offline_payment_service.setup_recurring(
            customer_data_with_email, payment_data
        )

    except Exception as e:
        print(
            f"Recurring payment setup failed and PaymentService raised an exception {e}"
        )

    try:
        error_payment_data = PaymentData(amount=100, source="tok_radarBlock")
        payment_service_email.process_transaction(
            customer_data_with_email, error_payment_data
        )
    except Exception as e:
        print(f"Payment failed and PaymentService raised an exception: {e}")

    # Set up recurrence
    recurring_payment_data = PaymentData(amount=100, source="pm_card_visa")
    payment_service_email.setup_recurring(
        customer_data_with_email, recurring_payment_data
    )

edgardo ponce

Hace 2 años

gracias

José Alexander Márquez Rivera

Hace un año

Thank you so much! 😉

Nicolas Potrich

Hace 2 años

Que los archivos after.py y before.py sean los mismos y sumado a los cambios que hace el profesor por fuera de las ediciones se hace muy dificil de poder seguir el curso. Por favor actualicen el curso para no tener que andar pausando y copiando el codigo a mano.

Bryann Andrei Valderrama Avila

Hace 2 años

Los archivos before.py y after.py son exactamente el mismo. Por favor, actualizarlos.

edgardo ponce

Hace 2 años

el curso no sigue una historia en el código. Arreglen las cosas, este curso es re interesante, pongan a alguien a chequear todo desde el inicio asi pueden ir corrigiendo todas las clases.

Yonalby Moya

Hace 2 años

Aunque solo entre a este curso para aprender pues estaba en mi ruta de aprendizaje en mi vida e visto un procesador de pago asi que aunque no entiendo mucho me estoy documentando y he ido aprendiendo de solid + programacion y otras cosas como el procesador de pago, ciertamente no es tan facil pero a veces quedo en el aire con algunas explicaciones. Bien agradezco el curso si puedes dar una explicacion asi sea en la descripcion algo mas detallado de algunos terminos seria excelente de resto gracias.

Vicente Millan

Hace 2 años

Entiendo que ustedes fomentan que la comunidad se apoye cuando ocurra un inconveniente, pero tampoco es para que nosotros les hagamos el trabajo gratis. Claramente, me refiero al problema de que los archivos before y after desde hace tres clases son los mismos, y esto dificulta que algunos alumnos puedan seguir la clase.

Cristian Barajas

Hace 2 años

el contenido de before y after es el mismo lo cual a mi me molesta un poco a la hora de querer practicar el refactorizado ya que no puedo tener el código que el modifico para la clase y debo pausar el video para copiar lo que el hizo

Laurens Sanchez

Hace un año

como hasta el minuto 1:48 se muestra el codigo sin segregacion de interfaces que deberia estar en el archivo de after.py donde vemos que el metodo OfflinePaymentProcessor esta obligado a tener metodos que no va usar. luego de alli empieza a explicar que los metodos RefunPayment y RecurrinPayment debe ser metodos abstractos para que puedan ser heredados unicamente donde se necesitan que en este caso seria en el metodo StripePaymentProcessor donde ya estarias usando la segregacion de interfaces.

Edgar Hernandez

Hace 2 años

Pregunta, la clase PaymentService no esta teniendo muchas responsabilidades?

Platzi

Hace 2 años

El código no indica que la clase PaymentService tenga muchas responsabilidades. Lo que sí explica es cómo las dependencias se inyectan para cumplir con el principio de inversión de dependencias.

Ulises Alejandro Mánica Caricio

Hace 2 años

Profe Eduardo... Una duda.... al final de la clase usted comenta que podemos usar "stripe_proccesor" como "refund_processor" y también como "recurring_proccesor"

DADO QUE

el stripe proccesor implementa las 3 interfaces... pero no entiendo porque

se da como parametro tipo

"refund_proccesor=stripe processor"

y en vez de esto no se crea una instancia de RefundPaymentProtocol()

tipo:

refund=RefundPaymentProtocol()

y esta se entrega como parametro

"refund_proccesor=refund"

Espero haberme dado a entender

Mario Alexander Vargas Celis

Hace 2 años

Para aplicar el **Principio de Segregación de Interfaces** (ISP) en el diseño de código, es clave asegurarse de que las clases no estén obligadas a depender de métodos que no usan. Esto se logra dividiendo interfaces grandes en varias interfaces más pequeñas y específicas, de manera que cada clase solo implemente los métodos que necesita.

### Ejemplo Práctico: Aplicación del ISP en Python

Imaginemos un sistema para manejar dispositivos electrónicos. Inicialmente, podríamos diseñar una interfaz llamada Device con métodos que representen varias funciones que un dispositivo podría tener, como encender, apagar y reproducir música. Sin embargo, no todos los dispositivos tienen las mismas capacidades. Un parlante puede reproducir música, pero una lámpara no.

#### Ejemplo Incorrecto: Violación del ISP

A continuación, un diseño que viola el ISP, en el que todos los dispositivos deben implementar todos los métodos, incluso si algunos no son relevantes.


class Device:

&#x20;   def turn\_on(self):

&#x20;       raise NotImplementedError



&#x20;   def turn\_off(self):

&#x20;       raise NotImplementedError



&#x20;   def play\_music(self):

&#x20;       raise NotImplementedError



class Speaker(Device):

&#x20;   def turn\_on(self):

&#x20;       print("Speaker is now on.")



&#x20;   def turn\_off(self):

&#x20;       print("Speaker is now off.")



&#x20;   def play\_music(self):

&#x20;       print("Playing music...")



class Lamp(Device):

&#x20;   def turn\_on(self):

&#x20;       print("Lamp is now on.")



&#x20;   def turn\_off(self):

&#x20;       print("Lamp is now off.")



&#x20;   def play\_music(self):

&#x20;       raise NotImplementedError("Lamps cannot play music.")

En este diseño, la clase Lamp debe implementar el método play\_music, a pesar de que una lámpara no reproduce música. Esto viola el ISP, ya que Lamp depende de un método que no le corresponde.

### Solución Correcta: Aplicación del ISP

Para aplicar el ISP, dividimos la interfaz Device en interfaces más pequeñas y específicas. Cada dispositivo solo implementará los métodos que realmente necesita.


from abc import ABC, abstractmethod



\# Interfaces específicas

class PowerDevice(ABC):

&#x20;   @abstractmethod

&#x20;   def turn\_on(self):

&#x20;       pass



&#x20;   @abstractmethod

&#x20;   def turn\_off(self):

&#x20;       pass



class MusicPlayer(ABC):

&#x20;   @abstractmethod

&#x20;   def play\_music(self):

&#x20;       pass



\# Clases que implementan las interfaces relevantes

class Speaker(PowerDevice, MusicPlayer):

&#x20;   def turn\_on(self):

&#x20;       print("Speaker is now on.")



&#x20;   def turn\_off(self):

&#x20;       print("Speaker is now off.")



&#x20;   def play\_music(self):

&#x20;       print("Playing music...")



class Lamp(PowerDevice):

&#x20;   def turn\_on(self):

&#x20;       print("Lamp is now on.")



&#x20;   def turn\_off(self):

&#x20;       print("Lamp is now off.")

### Explicación

- **Interfaz PowerDevice**: Esta interfaz define los métodos turn\_on y turn\_off, comunes a dispositivos que se pueden encender y apagar.

- **Interfaz MusicPlayer**: Define solo el método play\_music, exclusivo para dispositivos que pueden reproducir música.

- **Clase Speaker**: Implementa tanto PowerDevice como MusicPlayer, ya que puede encenderse y reproducir música.

- **Clase Lamp**: Implementa solo PowerDevice, ya que solo necesita los métodos de encendido y apagado.

### Uso del Código

Al aplicar el ISP, podemos usar Speaker y Lamp sin preocuparnos por métodos innecesarios o errores derivados de métodos irrelevantes.


\# Crear instancias de dispositivos

speaker = Speaker()

lamp = Lamp()



\# Usar métodos relevantes

speaker.turn\_on()     # Speaker is now on.

speaker.play\_music()  # Playing music...

speaker.turn\_off()    # Speaker is now off.



lamp.turn\_on()        # Lamp is now on.

lamp.turn\_off()       # Lamp is now off.

### Ventajas de Aplicar el ISP

1. **Código Más Limpio y Específico**: Cada clase implementa solo los métodos que necesita, eliminando métodos innecesarios.

2. **Mayor Cohesión y Mantenibilidad**: Las interfaces pequeñas y específicas facilitan el mantenimiento y la extensión del código.

3. **Mejora en la Legibilidad**: El código es más fácil de entender, ya que cada clase y cada interfaz tienen una responsabilidad clara.

### Conclusión

Aplicar el **Principio de Segregación de Interfaces** hace que el código sea más modular, fácil de mantener y específico. En este ejemplo, hemos logrado un diseño en el que cada dispositivo electrónico implementa solo los métodos que realmente utiliza, cumpliendo con el ISP y optimizando el sistema para agregar nuevos dispositivos sin generar dependencias innecesarias.

Gabriel de Jesus Chirinos Marcano

Hace 2 años

Entonces, ¿es recomendable usar las interfaces de los servicios como propiedades de un objeto, en vez de extender del propio Protocolo a la hora de crear la clase de dicho objeto?

Recursos

Aplicar el principio de segregación de interfaces en un procesador de pagos evita dependencias innecesarias y errores en tiempo de ejecución. Aquí verás cómo separar responsabilidades en protocolos específicos para cobros, reembolsos y recurrencias, con ejemplos prácticos usando Stripe y un procesador offline. La meta: clases que solo dependan de lo que realmente usan.

¿Cómo aplicar el principio de segregación de interfaces en un procesador de pagos?

El punto de partida es un procesador que ahora incluye dos métodos nuevos: reembolso y crear recurrencia. Además, se añadió un procesador de pagos offline que solo imprime el cobro, simulando efectivo. Al no tener pasarela, no puede reembolsar ni crear recurrencias, por lo que lanzar excepciones en esos métodos revela la violación al principio: una clase no debería depender de métodos que no puede implementar.

Se cambia el retorno de process transaction para no depender de un charge de Stripe. Ahora se define un resultado con: estatus exitoso o fallido, monto, ID de transacción y mensaje de la pasarela.
El procesador offline no soporta refund ni recurring: levantar excepciones lo hace evidente y guía la refactorización.

¿Qué problema revela el procesador de pagos offline?

Cuando el offline processor implementa métodos de reembolso y recurrencia que no puede cumplir, se obliga a los clientes a depender de funciones inútiles. Eso contraviene el principio de segregación de interfaces y dificulta el mantenimiento.

¿Qué protocolos separar: cobro, reembolso y recurrencia?

La solución es crear tres protocolos con un único propósito cada uno: uno para cobrar, otro para reembolsar y otro para recurrencia. Así, cada clase implementa solo lo que necesita.

from typing import Protocol, Optional

class PMProcessorProtocol(Protocol):
    def process_transaction(self, amount: int) -> dict: ...

class RefundPaymentProtocol(Protocol):
    def refund_payment(self, transaction_id: str) -> None: ...

class RecurringPaymentProtocol(Protocol):
    def create_recurring(self, amount: int, interval: str) -> None: ...

¿Qué cambios de diseño y código se realizan para Stripe y offline?

En Stripe, se implementan los tres métodos porque la pasarela soporta cobros, reembolsos y recurrencias. El procesador offline solo implementa cobros.

Stripe: hereda de los tres protocolos. Soporta todo el flujo.
Offline: hereda solo de PMProcessorProtocol. No tiene reembolso ni recurrencia.

class StripeProcessor(PMProcessorProtocol, RefundPaymentProtocol, RecurringPaymentProtocol):
    def process_transaction(self, amount: int) -> dict:
        # lógica real de Stripe
        return {"status": "success", "amount": amount}

    def refund_payment(self, transaction_id: str) -> None:
        # refund real en Stripe
        ...

    def create_recurring(self, amount: int, interval: str) -> None:
        # plan/recurrencia en Stripe
        ...

class OfflineProcessor(PMProcessorProtocol):
    def process_transaction(self, amount: int) -> dict:
        print(f"Cobrando en efectivo: {amount}")
        return {"status": "success", "amount": amount}

Nota: se menciona renombrar la clase base para cumplir un estándar: PM Processor Protocol. El objetivo es claridad y consistencia en nombres.

¿Cómo adaptar el servicio para refund y recurrencia opcionales?

Como los métodos de refund y recurring ya no están en el payment processor general, el service agrega dos atributos opcionales: refund_processor y recurring_processor. Se valida None antes de llamar y, de no existir, se lanza una excepción clara.

Agregar atributos opcionales: Optional[RefundPaymentProtocol] y Optional[RecurringPaymentProtocol].
Validar antes de ejecutar: si es None, lanzar Exception con mensajes explícitos.
Instanciación flexible: con Stripe, un solo procesador puede cubrir los tres flujos; con offline, no se pasan refund ni recurring.

class PaymentService:
    def __init__(
        self,
        payment_processor: PMProcessorProtocol,
        refund_processor: Optional[RefundPaymentProtocol] = None,
        recurring_processor: Optional[RecurringPaymentProtocol] = None,
    ) -> None:
        self.payment_processor = payment_processor
        self.refund_processor = refund_processor
        self.recurring_processor = recurring_processor

    def refund(self, transaction_id: str) -> None:
        if not self.refund_processor:
            raise Exception("this processor does not support refunds")
        self.refund_processor.refund_payment(transaction_id)

    def create_recurring(self, amount: int, interval: str) -> None:
        if not self.recurring_processor:
            raise Exception("This processor does not support recurrent")
        self.recurring_processor.create_recurring(amount, interval)

Instanciación con Stripe como único procesador para todo:

stripe = StripeProcessor()
service = PaymentService(
    payment_processor=stripe,
    refund_processor=stripe,
    recurring_processor=stripe,
)

Instanciación con offline solo para cobros en efectivo:

offline = OfflineProcessor()
service_cash = PaymentService(
    payment_processor=offline,
    # sin refund ni recurring para mantener la segregación
)

Palabras clave y habilidades puestas en práctica:

Principio de segregación de interfaces: evitar dependencias a métodos no usados.
Protocolos especializados: PMProcessorProtocol, RefundPaymentProtocol, RecurringPaymentProtocol.
Tipos opcionales y validación de None: robustez en el service.
Excepciones claras: mensajes orientados a capacidades del procesador.
Independencia de pasarela: retorno genérico en process transaction (estatus, monto, id, mensaje).

¿Qué otras mejoras aplicarías para reforzar la segregación de interfaces en este caso de uso de procesadores de pago? Comparte tus ideas y ejemplos en comentarios.

"""
features agregados: 
+ PaymentResponse: objeto
+ RefundPaymentProtocol: interfaz
+ PaymentProcessorProtocol:interfaz
+ RecurringPaymentProtocol: interfaz
+ StripePaymentProcessor: subclase (ejecuta demasiadas cosas)
+ OfflinePaymentProcessor: subclase
-------------------------------------------------------------
PREGUNTA: ¿Qué principio SOLID se estaba siguiendo en los métodos de recurrencia?
El principio S: single responsibity principle porque está dedicado úni-
camente a configurar los pagos con stripe y de la suscripción 
están delegadas a métodos auxiliares
"""

import os
from dataclasses import dataclass, field
from typing import Optional, Protocol
import uuid
import stripe
from dotenv import load_dotenv
from pydantic import BaseModel
from stripe.error import StripeError

_ = load_dotenv()


class ContactInfo(BaseModel):
    email: Optional[str] = None
    phone: Optional[str] = None


class CustomerData(BaseModel):
    name: str
    contact_info: ContactInfo
    customer_id: Optional[str] = None


class PaymentData(BaseModel):
    amount: int
    source: str


class PaymentResponse(BaseModel):
    status: str
    amount: int
    transaction_id: Optional[str] = None
    message: Optional[str] = None


class PaymentProcessor(Protocol): 
    """
    Protocol for processing payments, refunds, and recurring payments.
    # HEREDA A TODOS LOS PAYMENTPROCESOR
    This protocol defines the interface for payment processors. Implementations
    should provide methods for processing payments, refunds, and setting up recurring payments.
    """

    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse: ...
    
    def refund_payment(self, transaction_id:str) -> PaymentResponse: ...
    
    def setup_recurring_payment(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) ->PaymentResponse: ...


class StripePaymentProcessor(PaymentProcessor):
    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        try:
            charge = stripe.Charge.create(
                amount=payment_data.amount,
                currency="usd",
                source=payment_data.source,
                description="Charge for " + customer_data.name,
            )
            print("Payment successful")
            return PaymentResponse(
                status=charge["status"],
                amount=charge["amount"],
                transaction_id=charge["id"],
                message="Payment successful",
            )
        except StripeError as e:
            print("Payment failed:", e)
            return PaymentResponse(
                status="failed",
                amount=payment_data.amount,
                transaction_id=None,
                message=str(e),
            )

    def refund_payment(self, transaction_id: str) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        try:
            refund = stripe.Refund.create(charge=transaction_id)
            print("Refund successful")
            return PaymentResponse(
                status=refund["status"],
                amount=refund["amount"],
                transaction_id=refund["id"],
                message="Refund successful",
            )
        except StripeError as e:
            print("Refund failed:", e)
            return PaymentResponse(
                status="failed",
                amount=0,
                transaction_id=None,
                message=str(e),
            )

    def setup_recurring_payment(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        stripe.api_key = os.getenv("STRIPE_API_KEY")
        price_id = os.getenv("STRIPE_PRICE_ID", "")
        try:
            customer = self._get_or_create_customer(customer_data)

            payment_method = self._attach_payment_method(
                customer.id, payment_data.source
            )

            self._set_default_payment_method(customer.id, payment_method.id)

            subscription = stripe.Subscription.create(
                customer=customer.id,
                items=[
                    {"price": price_id},
                ],
                expand=["latest_invoice.payment_intent"],
            )

            print("Recurring payment setup successful")
            amount = subscription["items"]["data"][0]["price"]["unit_amount"]
            return PaymentResponse(
                status=subscription["status"],
                amount=amount,
                transaction_id=subscription["id"],
                message="Recurring payment setup successful",
            )
        except StripeError as e:
            print("Recurring payment setup failed:", e)
            return PaymentResponse(
                status="failed",
                amount=0,
                transaction_id=None,
                message=str(e),
            )

    def _get_or_create_customer(
        self, customer_data: CustomerData
    ) -> stripe.Customer:
        """
        Creates a new customer in Stripe or retrieves an existing one.
        """
        if customer_data.customer_id:
            customer = stripe.Customer.retrieve(customer_data.customer_id)
            print(f"Customer retrieved: {customer.id}")
        else:
            if not customer_data.contact_info.email:
                raise ValueError("Email required for subscriptions")
            customer = stripe.Customer.create(
                name=customer_data.name, email=customer_data.contact_info.email
            )
            print(f"Customer created: {customer.id}")
        return customer

    def _attach_payment_method(
        self, customer_id: str, payment_source: str
    ) -> stripe.PaymentMethod:
        """
        Attaches a payment method to a customer.
        """
        payment_method = stripe.PaymentMethod.retrieve(payment_source)
        stripe.PaymentMethod.attach(
            payment_method.id,
            customer=customer_id,
        )
        print(
            f"Payment method {payment_method.id} attached to customer {customer_id}"
        )
        return payment_method

    def _set_default_payment_method(
        self, customer_id: str, payment_method_id: str
    ) -> None:
        """
        Sets the default payment method for a customer.
        """
        stripe.Customer.modify(
            customer_id,
            invoice_settings={
                "default_payment_method": payment_method_id,
            },
        )
        print(f"Default payment method set for customer {customer_id}")


class OfflinePaymentProcessor(PaymentProcessor):
    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        print("Processing offline payment for", customer_data.name)
        return PaymentResponse(
            status="success",
            amount=payment_data.amount,
            transaction_id=str(uuid.uuid4()),
            message="Offline payment success",
        )
    """
    los siguientes métodos levantan errores porque no se pueden hacer reembolsos ni recurrencias a efectivo
    se incumple el principio de segregación de interfaces porque una clase no debería depender 
    de clases que no puede implementar
    """
    def refund_payment(self, transaction_id: str) -> PaymentResponse:
        print("refunds aren't supported in OfflinePaymentOricessor.")
        raise NotImplementedError("Refunds not supported in offline processor.")
    
    def setup_recurring_payment(self, customer_data: CustomerData, payment_data: PaymentData
    ) ->PaymentResponse:
        print("recurring payments aren't supported in OfflinePaymentOricessor.")
        raise NotImplementedError("Refunds not supported in offline processor.")
    

class Notifier(Protocol):
    """
    Protocol for sending notifications.

    This protocol defines the interface for notifiers. Implementations
    should provide a method `send_confirmation` that sends a confirmation
    to the customer.
    """

    def send_confirmation(self, customer_data: CustomerData): ...


class EmailNotifier:
    def send_confirmation(self, customer_data: CustomerData):
        from email.mime.text import MIMEText

        if not customer_data.contact_info.email:
            raise ValueError("Email address is requiered to send an email")

        msg = MIMEText("Thank you for your payment.")
        msg["Subject"] = "Payment Confirmation"
        msg["From"] = "no-reply@example.com"
        msg["To"] = customer_data.contact_info.email

        print("Email sent to", customer_data.contact_info.email)


@dataclass
class SMSNotifier:
    gateway: str

    def send_confirmation(self, customer_data: CustomerData):
        phone_number = customer_data.contact_info.phone
        if not phone_number:
            print("No phone number provided")
            return
        print(
            f"SMS sent to {phone_number} via {self.gateway}: Thank you for your payment."
        )


class TransactionLogger:
    def log_transaction(
        self,
        customer_data: CustomerData,
        payment_data: PaymentData,
        payment_response: PaymentResponse,
    ):
        with open("transactions.log", "a") as log_file:
            log_file.write(
                f"{customer_data.name} paid {payment_data.amount}\n"
            )
            log_file.write(f"Payment status: {payment_response.status}\n")
            if payment_response.transaction_id:
                log_file.write(
                    f"Transaction ID: {payment_response.transaction_id}\n"
                )
            log_file.write(f"Message: {payment_response.message}\n")

    def log_refund(
        self, transaction_id: str, refund_response: PaymentResponse
    ):
        with open("transactions.log", "a") as log_file:
            log_file.write(
                f"Refund processed for transaction {transaction_id}\n"
            )
            log_file.write(f"Refund status: {refund_response.status}\n")
            log_file.write(f"Message: {refund_response.message}\n")


class CustomerValidator:
    def validate(self, customer_data: CustomerData):
        if not customer_data.name:
            print("Invalid customer data: missing name")
            raise ValueError("Invalid customer data: missing name")
        if not customer_data.contact_info:
            print("Invalid customer data: missing contact info")
            raise ValueError("Invalid customer data: missing contact info")
        if not (
            customer_data.contact_info.email
            or customer_data.contact_info.phone
        ):
            print("Invalid customer data: missing email and phone")
            raise ValueError("Invalid customer data: missing email and phone")


class PaymentDataValidator:
    def validate(self, payment_data: PaymentData):
        if not payment_data.source:
            print("Invalid payment data: missing source")
            raise ValueError("Invalid payment data: missing source")
        if payment_data.amount <= 0:
            print("Invalid payment data: amount must be positive")
            raise ValueError("Invalid payment data: amount must be positive")


@dataclass
class PaymentService:
    payment_processor: PaymentProcessor
    notifier: Notifier
    customer_validator: CustomerValidator = field(
        default_factory=CustomerValidator
    )
    payment_validator: PaymentDataValidator = field(
        default_factory=PaymentDataValidator
    )
    logger: TransactionLogger = field(default_factory=TransactionLogger)

    def process_transaction(
        self, customer_data: CustomerData, payment_data: PaymentData
    ) -> PaymentResponse:
        self.customer_validator.validate(customer_data)
        self.payment_validator.validate(payment_data)
        payment_response = self.payment_processor.process_transaction(
            customer_data, payment_data
        )
        self.notifier.send_confirmation(customer_data)
        self.logger.log_transaction(
            customer_data, payment_data, payment_response
        )
        return payment_response

    def process_refund(self, transaction_id: str):
        refund_response = self.payment_processor.refund_payment(transaction_id)
        self.logger.log_refund(transaction_id, refund_response)
        return refund_response

    def setup_recurring(
        self, customer_data: CustomerData, payment_data: PaymentData
    ):
        recurring_response = self.payment_processor.setup_recurring_payment(
            customer_data, payment_data
        )
        self.logger.log_transaction(
            customer_data, payment_data, recurring_response
        )
        return recurring_response


if __name__ == "__main__":
    # Set up the payment processors
    stripe_processor = StripePaymentProcessor()
    offline_processor = OfflinePaymentProcessor()

    # Set up the customer data and payment data
    customer_data_with_email = CustomerData(
        name="John Doe", contact_info=ContactInfo(email="john@example.com")
    )
    customer_data_with_phone = CustomerData(
        name="Jane Doe", contact_info=ContactInfo(phone="1234567890")
    )

    # Set up the payment data
    payment_data = PaymentData(amount=100, source="tok_visa")

    # Set up the notifiers
    email_notifier = EmailNotifier()

    sms_gateway = "YourSMSService"
    sms_notifier = SMSNotifier(sms_gateway)

    # # Using Stripe processor with email notifier
    payment_service_email = PaymentService(stripe_processor, email_notifier)
    payment_service_email.process_transaction(customer_data_with_email, payment_data)

    # Using Stripe processor with SMS notifier
    payment_service_sms = PaymentService(stripe_processor, sms_notifier)
    sms_payment_response = payment_service_sms.process_transaction(customer_data_with_phone, payment_data)

    #Using strupe processor with SMS notifier
    payment_service_sms = PaymentService(stripe_processor, sms_notifier)
    sms_payment_response = payment_service_sms.process_transaction(customer_data_with_phone, payment_data)


    # Example of processing a refund using Stripe processor
    transaction_id_to_refund = sms_payment_response.transaction_id
    if transaction_id_to_refund:
        payment_service_email.process_refund(transaction_id_to_refund)

    # Using offline processor with email notifier
    offline_payment_service = PaymentService(offline_processor, email_notifier)
    offline_payment_response = offline_payment_service.process_transaction(
        customer_data_with_email, payment_data
    )

    # Attempt to refund using offline processor (will fail)
    try:
        if offline_payment_response.transaction_id:
            offline_payment_service.process_refund(
                offline_payment_response.transaction_id
            )
    except Exception as e:
        print(f"Refund failed and PaymentService raised an exception: {e}")

    # Attempt to set up recurring payment using offline processor (will fail)
    try:
        offline_payment_service.setup_recurring(
            customer_data_with_email, payment_data
        )

    except Exception as e:
        print(
            f"Recurring payment setup failed and PaymentService raised an exception {e}"
        )

    try:
        error_payment_data = PaymentData(amount=100, source="tok_radarBlock")
        payment_service_email.process_transaction(
            customer_data_with_email, error_payment_data
        )
    except Exception as e:
        print(f"Payment failed and PaymentService raised an exception: {e}")

    # Set up recurrence
    recurring_payment_data = PaymentData(amount=100, source="pm_card_visa")
    payment_service_email.setup_recurring(
        customer_data_with_email, recurring_payment_data
    )

class Device:

&#x20;   def turn\_on(self):

&#x20;       raise NotImplementedError

&#x20;   def turn\_off(self):

&#x20;       raise NotImplementedError

&#x20;   def play\_music(self):

&#x20;       raise NotImplementedError

class Speaker(Device):

&#x20;   def turn\_on(self):

&#x20;       print("Speaker is now on.")

&#x20;   def turn\_off(self):

&#x20;       print("Speaker is now off.")

&#x20;   def play\_music(self):

&#x20;       print("Playing music...")

class Lamp(Device):

&#x20;   def turn\_on(self):

&#x20;       print("Lamp is now on.")

&#x20;   def turn\_off(self):

&#x20;       print("Lamp is now off.")

&#x20;   def play\_music(self):

&#x20;       raise NotImplementedError("Lamps cannot play music.")

from abc import ABC, abstractmethod

\# Interfaces específicas

class PowerDevice(ABC):

&#x20;   @abstractmethod

&#x20;   def turn\_on(self):

&#x20;       pass

&#x20;   @abstractmethod

&#x20;   def turn\_off(self):

&#x20;       pass

class MusicPlayer(ABC):

&#x20;   @abstractmethod

&#x20;   def play\_music(self):

&#x20;       pass

\# Clases que implementan las interfaces relevantes

class Speaker(PowerDevice, MusicPlayer):

&#x20;   def turn\_on(self):

&#x20;       print("Speaker is now on.")

&#x20;   def turn\_off(self):

&#x20;       print("Speaker is now off.")

&#x20;   def play\_music(self):

&#x20;       print("Playing music...")

class Lamp(PowerDevice):

&#x20;   def turn\_on(self):

&#x20;       print("Lamp is now on.")

&#x20;   def turn\_off(self):

&#x20;       print("Lamp is now off.")

\# Crear instancias de dispositivos

speaker = Speaker()

lamp = Lamp()

\# Usar métodos relevantes

speaker.turn\_on()     # Speaker is now on.

speaker.play\_music()  # Playing music...

speaker.turn\_off()    # Speaker is now off.

lamp.turn\_on()        # Lamp is now on.

lamp.turn\_off()       # Lamp is now off.

Resumen

¿Cómo aplicar el principio de segregación de interfaces en un procesador de pagos?

Se cambia el retorno de process transaction para no depender de un charge de Stripe. Ahora se define un resultado con: estatus exitoso o fallido, monto, ID de transacción y mensaje de la pasarela.
El procesador offline no soporta refund ni recurring: levantar excepciones lo hace evidente y guía la refactorización.

¿Qué problema revela el procesador de pagos offline?

¿Qué protocolos separar: cobro, reembolso y recurrencia?

La solución es crear tres protocolos con un único propósito cada uno: uno para cobrar, otro para reembolsar y otro para recurrencia. Así, cada clase implementa solo lo que necesita.

from typing import Protocol, Optional

class PMProcessorProtocol(Protocol):
    def process_transaction(self, amount: int) -> dict: ...

class RefundPaymentProtocol(Protocol):
    def refund_payment(self, transaction_id: str) -> None: ...

class RecurringPaymentProtocol(Protocol):
    def create_recurring(self, amount: int, interval: str) -> None: ...

¿Qué cambios de diseño y código se realizan para Stripe y offline?

En Stripe, se implementan los tres métodos porque la pasarela soporta cobros, reembolsos y recurrencias. El procesador offline solo implementa cobros.

Stripe: hereda de los tres protocolos. Soporta todo el flujo.
Offline: hereda solo de PMProcessorProtocol. No tiene reembolso ni recurrencia.

class StripeProcessor(PMProcessorProtocol, RefundPaymentProtocol, RecurringPaymentProtocol):
    def process_transaction(self, amount: int) -> dict:
        # lógica real de Stripe
        return {"status": "success", "amount": amount}

    def refund_payment(self, transaction_id: str) -> None:
        # refund real en Stripe
        ...

    def create_recurring(self, amount: int, interval: str) -> None:
        # plan/recurrencia en Stripe
        ...

class OfflineProcessor(PMProcessorProtocol):
    def process_transaction(self, amount: int) -> dict:
        print(f"Cobrando en efectivo: {amount}")
        return {"status": "success", "amount": amount}

Nota: se menciona renombrar la clase base para cumplir un estándar: PM Processor Protocol. El objetivo es claridad y consistencia en nombres.

¿Cómo adaptar el servicio para refund y recurrencia opcionales?

Agregar atributos opcionales: Optional[RefundPaymentProtocol] y Optional[RecurringPaymentProtocol].
Validar antes de ejecutar: si es None, lanzar Exception con mensajes explícitos.
Instanciación flexible: con Stripe, un solo procesador puede cubrir los tres flujos; con offline, no se pasan refund ni recurring.

class PaymentService:
    def __init__(
        self,
        payment_processor: PMProcessorProtocol,
        refund_processor: Optional[RefundPaymentProtocol] = None,
        recurring_processor: Optional[RecurringPaymentProtocol] = None,
    ) -> None:
        self.payment_processor = payment_processor
        self.refund_processor = refund_processor
        self.recurring_processor = recurring_processor

    def refund(self, transaction_id: str) -> None:
        if not self.refund_processor:
            raise Exception("this processor does not support refunds")
        self.refund_processor.refund_payment(transaction_id)

    def create_recurring(self, amount: int, interval: str) -> None:
        if not self.recurring_processor:
            raise Exception("This processor does not support recurrent")
        self.recurring_processor.create_recurring(amount, interval)

Instanciación con Stripe como único procesador para todo:

stripe = StripeProcessor()
service = PaymentService(
    payment_processor=stripe,
    refund_processor=stripe,
    recurring_processor=stripe,
)

Instanciación con offline solo para cobros en efectivo:

offline = OfflineProcessor()
service_cash = PaymentService(
    payment_processor=offline,
    # sin refund ni recurring para mantener la segregación
)

Palabras clave y habilidades puestas en práctica:

Principio de segregación de interfaces: evitar dependencias a métodos no usados.
Protocolos especializados: PMProcessorProtocol, RefundPaymentProtocol, RecurringPaymentProtocol.
Tipos opcionales y validación de None: robustez en el service.
Excepciones claras: mensajes orientados a capacidades del procesador.
Independencia de pasarela: retorno genérico en process transaction (estatus, monto, id, mensaje).

¿Qué otras mejoras aplicarías para reforzar la segregación de interfaces en este caso de uso de procesadores de pago? Comparte tus ideas y ejemplos en comentarios.

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

Segregación de interfaces en procesadores de pagos

Progreso del curso

Patrones de Diseño y Principios SOLID

Principios SOLID

Reestructuración del proyecto

Patrones de Diseño

Recursos

Recursos