Stripe usa Checkout Sessions. PayPal usa la API de Orders con tokens OAuth2. Hub2 envía un USSD push al teléfono del cliente y espera. PawaPay redirige a una página de pago alojada. Cada proveedor tiene su propio esquema de autenticación, su propia estructura de API, su propio formato de webhook y su propia definición de "pago iniciado". Sin embargo, en 0fee.dev, iniciar un pago a través de cualquiera de estos proveedores se ve exactamente igual para el código que lo llama. Esta es la historia del patrón adaptador que lo hace posible.
El problema: fragmentación de proveedores de pago
Un orquestador de pagos se sitúa entre comerciantes y proveedores de pago. Toda su propuesta de valor es que el comerciante se integra una vez y obtiene acceso a cada proveedor. Pero "cada proveedor" significa lidiar con APIs radicalmente diferentes:
| Proveedor | Método de auth | Flujo de pago | Notificación de estado |
|---|---|---|---|
| Stripe | Token bearer (clave secreta) | Crear Checkout Session, redirigir a página alojada de Stripe | Webhook (firma verificada) |
| PayPal | Credenciales de cliente OAuth2 | Crear Order, redirigir a URL de aprobación de PayPal | Webhook + polling |
| Hub2 | Clave API + clave de suscripción | POST para iniciar, USSD push enviado al teléfono | Callback de webhook |
| PawaPay | Token bearer | POST para iniciar, redirigir a página alojada | Callback de webhook |
| PaiementPro | Clave API en el cuerpo | POST para iniciar, redirigir a página de pago | URL de callback |
| BUI | Clave API + HMAC | POST para iniciar, OTP o redirección Wave | Webhook con HMAC |
| Test | Ninguno | Resultado instantáneo basado en monto mágico | Ninguno (síncrono) |
Sin una capa de abstracción, el código de iniciación de pago sería un enorme switch, el manejador de webhooks sería otro, y cada nuevo proveedor requeriría cambios en docenas de lugares.
La clase abstracta BasePayinProvider
La solución fue una clase base abstracta en Python que define el contrato que cada adaptador de proveedor debe cumplir. Aquí está la clase base real de backend/providers/base.py:
pythonfrom abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class InitPaymentResult:
"""Resultado de iniciar un pago con un proveedor."""
provider_ref: str
status: str # "pending", "redirect", "ussd_push", "failed"
redirect_url: Optional[str] = None
ussd_code: Optional[str] = None
instructions: Optional[str] = None
raw_response: dict = field(default_factory=dict)
class BasePayinProvider(ABC):
"""Clase base abstracta para todos los proveedores de pago."""
provider_id: str
provider_name: str
supported_countries: list[str]
supported_methods: list[str]
def __init__(self, credentials: dict):
"""Inicializar con credenciales de proveedor descifradas."""
self.credentials = credentials
@abstractmethod
async def initiate_payment(self, data: dict) -> InitPaymentResult:
"""
Iniciar un pago con este proveedor.
Args:
data: Dict conteniendo monto, moneda, info del cliente,
método de pago, callback_url, etc.
Returns:
InitPaymentResult con provider_ref e información del flujo.
"""
pass
@abstractmethod
async def get_status(self, provider_ref: str) -> dict:
"""
Verificar estado del pago con el proveedor.
Returns:
Dict con estado, provider_ref e información adicional.
"""
pass
@abstractmethod
async def handle_webhook(
self, payload: dict, headers: dict
) -> dict:
"""
Procesar un webhook entrante de este proveedor.
Returns:
Dict con transaction_id, estado y datos normalizados.
"""
pass
async def refund(
self, provider_ref: str, amount: Optional[int] = None
) -> dict:
"""Reembolsar un pago. Opcional -- no todos los proveedores lo soportan."""
raise NotImplementedError(
f"{self.provider_name} does not support refunds"
)
async def cancel_payment(self, provider_ref: str) -> dict:
"""Cancelar un pago pendiente. Opcional."""
raise NotImplementedError(
f"{self.provider_name} does not support cancellation"
)
async def validate_credentials(self) -> bool:
"""Validar que las credenciales almacenadas son correctas."""
raise NotImplementedError(
f"{self.provider_name} does not support credential validation"
)Las decisiones de diseño incorporadas en esta clase merecen examinarse:
- Tres métodos obligatorios, tres opcionales. Cada proveedor debe implementar
initiate_payment,get_statusyhandle_webhook. Reembolsos, cancelaciones y validación de credenciales son opcionales porque no todos los proveedores los soportan.
- Credenciales pasadas en la construcción. La instancia del proveedor recibe credenciales descifradas cuando se crea. Esto significa que la capa de cifrado es completamente invisible para las implementaciones de proveedores.
InitPaymentResultes un dataclass, no un dict. Usar un dataclass tipado obliga a cada proveedor a devolver la misma estructura. El campostatuses particularmente importante -- le dice al llamador qué hacer a continuación.
- El parámetro
dataes un dict, no un modelo. Esta fue una compensación deliberada. Diferentes proveedores necesitan diferentes campos (algunos necesitan un número de teléfono, otros un email, otros una URL de redirección). Un dict con claves documentadas fue más flexible que un modelo rígido.
Los tipos de estado de InitPaymentResult
El campo status en InitPaymentResult es la clave para manejar diferentes flujos de pago de manera uniforme:
| Estado | Significado | Qué hace el llamador |
|---|---|---|
pending | El pago se está procesando (USSD push enviado, esperando confirmación) | Consultar actualizaciones de estado |
redirect | El cliente debe ser redirigido para completar el pago | Devolver URL de redirección al frontend |
ussd_push | Prompt USSD enviado al teléfono del cliente | Mostrar mensaje "revisa tu teléfono", consultar estado |
failed | La iniciación del pago falló | Devolver error al comerciante |
Esta abstracción es lo que permite al orquestador manejar Stripe (redirigir a página de checkout), Hub2 (USSD push al teléfono) y PawaPay (redirigir a página alojada) a través del mismo camino de código.
Cómo Stripe implementa la interfaz
El adaptador de Stripe envuelve la API de Checkout Sessions:
pythonimport httpx
from providers.base import BasePayinProvider, InitPaymentResult
class StripeProvider(BasePayinProvider):
provider_id = "stripe"
provider_name = "Stripe"
supported_countries = ["GLOBAL"]
supported_methods = ["card"]
def __init__(self, credentials: dict):
super().__init__(credentials)
self.api_key = credentials.get("api_key", "")
self.webhook_secret = credentials.get("webhook_secret", "")
self.base_url = "https://api.stripe.com/v1"
async def initiate_payment(self, data: dict) -> InitPaymentResult:
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/checkout/sessions",
auth=(self.api_key, ""),
data={
"payment_method_types[]": "card",
"mode": "payment",
"line_items[0][price_data][currency]":
data["currency"].lower(),
"line_items[0][price_data][unit_amount]":
data["amount"],
"line_items[0][price_data][product_data][name]":
data.get("description", "Payment"),
"line_items[0][quantity]": 1,
"success_url": data.get("return_url", "")
+ "?status=success",
"cancel_url": data.get("return_url", "")
+ "?status=cancelled",
"client_reference_id": data.get("reference", ""),
"customer_email": data.get("customer", {})
.get("email", ""),
"metadata[zerofee_txn]": data.get("transaction_id", ""),
}
)
result = response.json()
if response.status_code == 200:
return InitPaymentResult(
provider_ref=result["id"],
status="redirect",
redirect_url=result["url"],
raw_response=result
)
else:
return InitPaymentResult(
provider_ref="",
status="failed",
raw_response=result
)
async def get_status(self, provider_ref: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/checkout/sessions/{provider_ref}",
auth=(self.api_key, "")
)
session = response.json()
status_map = {
"complete": "completed",
"expired": "expired",
"open": "pending",
}
return {
"status": status_map.get(
session.get("status"), "pending"
),
"provider_ref": provider_ref,
"payment_intent": session.get("payment_intent"),
}
async def handle_webhook(
self, payload: dict, headers: dict
) -> dict:
# Verificar firma de Stripe
import hmac, hashlib
signature = headers.get("stripe-signature", "")
# ... lógica de verificación de firma ...
event_type = payload.get("type", "")
if event_type == "checkout.session.completed":
session = payload["data"]["object"]
return {
"transaction_id": session["metadata"]
.get("zerofee_txn"),
"status": "completed",
"provider_ref": session["id"],
"amount_received": session
.get("amount_total", 0),
}
return {"status": "ignored"}Observaciones clave: Stripe devuelve status="redirect" con una redirect_url. El llamador envía esta URL al frontend del comerciante, que redirige al cliente a la página de checkout alojada de Stripe. El manejador de webhook normaliza el formato de evento de Stripe al formato estándar de 0fee.dev.
Cómo Hub2 implementa la misma interfaz
Hub2 sirve al África francófona con pagos USSD push. La misma interfaz, flujo completamente diferente:
pythonimport httpx
from providers.base import BasePayinProvider, InitPaymentResult
class Hub2Provider(BasePayinProvider):
provider_id = "hub2"
provider_name = "Hub2"
supported_countries = [
"CI", "SN", "BJ", "BF", "CM", "ML", "TG", "GN"
]
supported_methods = ["orange_money", "mtn", "wave", "moov"]
def __init__(self, credentials: dict):
super().__init__(credentials)
self.api_key = credentials.get("api_key", "")
self.subscription_key = credentials.get(
"subscription_key", ""
)
self.base_url = "https://api.hub2.io/v1"
async def initiate_payment(self, data: dict) -> InitPaymentResult:
# Hub2 usa códigos de operador, no nombres de métodos
operator_map = {
"orange_money": "Orange",
"mtn": "MTN",
"wave": "Wave",
"moov": "Moov",
}
operator = operator_map.get(
data.get("payment_method_detail", ""), "Orange"
)
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/payments",
headers={
"Authorization": f"Bearer {self.api_key}",
"Ocp-Apim-Subscription-Key":
self.subscription_key,
},
json={
"amount": data["amount"],
"currency": data.get("currency", "XOF"),
"customer": {
"phone": data["customer"]["phone"],
"name": data.get("customer", {})
.get("name", ""),
},
"operator": operator,
"country": data.get("country", "CI"),
"callback_url": data.get("callback_url", ""),
"metadata": {
"zerofee_txn": data
.get("transaction_id", "")
},
}
)
result = response.json()
if response.status_code in (200, 201):
return InitPaymentResult(
provider_ref=result.get("id", ""),
status="ussd_push",
instructions=(
"A USSD prompt has been sent to your "
"phone. Please confirm the payment."
),
raw_response=result
)
else:
return InitPaymentResult(
provider_ref="",
status="failed",
raw_response=result
)El contraste es marcado. Donde Stripe devuelve status="redirect", Hub2 devuelve status="ussd_push". Donde Stripe proporciona una redirect_url, Hub2 proporciona instructions indicando al cliente que revise su teléfono. Pero al código que llama no le importa:
python# En el manejador de ruta de pago -- mismo código para cada proveedor
result = await provider.initiate_payment(payment_data)
if result.status == "redirect":
# Enviar URL de redirección al frontend
return {"payment_flow": "redirect", "url": result.redirect_url}
elif result.status == "ussd_push":
# Decir al frontend que muestre "revisa tu teléfono"
return {"payment_flow": "ussd_push", "message": result.instructions}
elif result.status == "pending":
# El pago se está procesando
return {"payment_flow": "pending"}
elif result.status == "failed":
# Devolver error
return {"error": "Payment initiation failed"}El registro de proveedores
Los adaptadores de proveedores no se instancian directamente. En su lugar, un registro los gestiona:
pythonclass ProviderRegistry:
"""Registro para gestionar instancias de proveedores de pago."""
_providers: dict[str, type[BasePayinProvider]] = {}
_instances: dict[str, BasePayinProvider] = {}
def register(
self, provider_id: str, provider_class: type[BasePayinProvider]
):
"""Registrar una clase de proveedor."""
self._providers[provider_id] = provider_class
def get_instance(
self, provider_id: str, credentials: dict,
app_id: str = ""
) -> BasePayinProvider:
"""
Obtener o crear una instancia de proveedor.
Las instancias se cachean por provider_id + app_id.
"""
cache_key = f"{provider_id}:{app_id}"
if cache_key not in self._instances:
provider_class = self._providers.get(provider_id)
if not provider_class:
raise ValueError(
f"Unknown provider: {provider_id}"
)
self._instances[cache_key] = provider_class(credentials)
return self._instances[cache_key]
def list_providers(self) -> list[dict]:
"""Listar todos los proveedores registrados."""
return [
{
"id": pid,
"name": cls.provider_name,
"countries": cls.supported_countries,
"methods": cls.supported_methods,
}
for pid, cls in self._providers.items()
]
# Instancia global del registro
provider_registry = ProviderRegistry()
# Registrar todos los proveedores al cargar el módulo
from providers.stripe.provider import StripeProvider
from providers.paypal.provider import PayPalProvider
from providers.hub2.provider import Hub2Provider
from providers.pawapay.provider import PawaPayProvider
from providers.test.provider import TestProvider
provider_registry.register("stripe", StripeProvider)
provider_registry.register("paypal", PayPalProvider)
provider_registry.register("hub2", Hub2Provider)
provider_registry.register("pawapay", PawaPayProvider)
provider_registry.register("test", TestProvider)El registro proporciona dos capacidades importantes:
- Caché de instancias. Las instancias de proveedores se cachean por
provider_id:app_id, así las credenciales se cargan una vez y se reutilizan entre solicitudes. Esto evita re-descifrar credenciales en cada pago.
- Gestión dinámica. Nuevos proveedores pueden registrarse en tiempo de ejecución sin reiniciar la aplicación. Cuando BUI y PaiementPro se agregaron en la Sesión 002, simplemente se registraron en el mismo registro.
¿Por qué no usar un sistema de plugins?
Algunos orquestadores de pagos usan arquitecturas de plugins con carga en caliente, archivos de configuración y descubrimiento dinámico. Elegimos un enfoque más simple:
- Los proveedores son módulos Python. Agregar un nuevo proveedor significa crear un nuevo directorio con un archivo
provider.pyy registrarlo en el registro. Sin archivos de configuración, sin manifiestos de plugin, sin mecanismos de descubrimiento. - Se mantiene la seguridad de tipos. Como las clases de proveedores heredan de
BasePayinProvider, los IDEs y verificadores de tipos pueden confirmar que cada método requerido está implementado. - Las pruebas son directas. El proveedor Test implementa la misma interfaz con resultados deterministas, haciendo trivial probar flujos de pago sin hacer mocks de APIs externas.
El patrón adaptador no es novedoso. Es un patrón de diseño de libro de texto. Pero en el contexto de un orquestador de pagos, donde la complejidad de la integración de proveedores es el desafío central, aplicarlo consistentemente a través de cada proveedor es lo que hace el sistema manejable. Cuando 0fee.dev creció de 5 proveedores a 7 en la Sesión 002, ningún código existente fue modificado -- solo se agregaron y registraron nuevos módulos adaptadores.
El patrón en la práctica: agregar un nuevo proveedor
Cuando se agregó BUI en la Sesión 002, el proceso fue:
- Crear
backend/providers/bui/__init__.pyybackend/providers/bui/provider.py. - Implementar
BuiProvider(BasePayinProvider)con los tres métodos requeridos. - Agregar una línea al registro:
provider_registry.register("bui", BuiProvider).
Sin cambios en la ruta de pago. Sin cambios en el manejador de webhook. Sin cambios en el motor de enrutamiento. El nuevo proveedor estuvo inmediatamente disponible para cualquier app que configurara credenciales de BUI.
Este es el poder del patrón adaptador en un orquestador de pagos: la interfaz absorbe la complejidad de cada proveedor, y el resto del sistema nunca necesita saber si un pago pasó por la API de Checkout Sessions de Stripe o el mecanismo de USSD push de Hub2. Una interfaz, cada sistema de pago.
Este artículo es parte de la serie "Cómo construimos 0fee.dev". 0fee.dev es un orquestador de pagos que cubre más de 53 proveedores en más de 200 países, construido por Juste A. GNIMAVO y Claude desde Abiyán sin ningún ingeniero humano. Sigue la serie para conocer la historia completa de construcción.