Stripe uses Checkout Sessions. PayPal uses the Orders API with OAuth2 tokens. Hub2 sends a USSD push to the customer's phone and waits. PawaPay redirects to a hosted payment page. Each provider has its own authentication scheme, its own API structure, its own webhook format, and its own definition of "payment initiated." Yet in 0fee.dev, initiating a payment through any of these providers looks exactly the same to the calling code. This is the story of the adapter pattern that makes it possible.
The Problem: Payment Provider Fragmentation
A payment orchestrator sits between merchants and payment providers. Its entire value proposition is that the merchant integrates once and gains access to every provider. But "every provider" means dealing with wildly different APIs:
| Provider | Auth Method | Payment Flow | Status Notification |
|---|---|---|---|
| Stripe | Bearer token (secret key) | Create Checkout Session, redirect to Stripe-hosted page | Webhook (signature verified) |
| PayPal | OAuth2 client credentials | Create Order, redirect to PayPal approval URL | Webhook + polling |
| Hub2 | API key + subscription key | POST to initiate, USSD push sent to phone | Webhook callback |
| PawaPay | Bearer token | POST to initiate, redirect to hosted page | Webhook callback |
| PaiementPro | API key in body | POST to initiate, redirect to payment page | Callback URL |
| BUI | API key + HMAC | POST to initiate, OTP or Wave redirect | Webhook with HMAC |
| Test | None | Instant result based on magic amount | None (synchronous) |
Without an abstraction layer, the payment initiation code would be a massive switch statement, the webhook handler would be another, and every new provider would require changes in dozens of places.
The BasePayinProvider Abstract Class
The solution was a Python abstract base class that defines the contract every provider adapter must fulfill. Here is the actual base class from backend/providers/base.py:
pythonfrom abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class InitPaymentResult:
"""Result of initiating a payment with a provider."""
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):
"""Abstract base class for all payment providers."""
provider_id: str
provider_name: str
supported_countries: list[str]
supported_methods: list[str]
def __init__(self, credentials: dict):
"""Initialize with decrypted provider credentials."""
self.credentials = credentials
@abstractmethod
async def initiate_payment(self, data: dict) -> InitPaymentResult:
"""
Initiate a payment with this provider.
Args:
data: Dict containing amount, currency, customer info,
payment_method, callback_url, etc.
Returns:
InitPaymentResult with provider_ref and flow information.
"""
pass
@abstractmethod
async def get_status(self, provider_ref: str) -> dict:
"""
Check payment status with the provider.
Returns:
Dict with status, provider_ref, and any additional info.
"""
pass
@abstractmethod
async def handle_webhook(
self, payload: dict, headers: dict
) -> dict:
"""
Process an incoming webhook from this provider.
Returns:
Dict with transaction_id, status, and normalized data.
"""
pass
async def refund(
self, provider_ref: str, amount: Optional[int] = None
) -> dict:
"""Refund a payment. Optional -- not all providers support it."""
raise NotImplementedError(
f"{self.provider_name} does not support refunds"
)
async def cancel_payment(self, provider_ref: str) -> dict:
"""Cancel a pending payment. Optional."""
raise NotImplementedError(
f"{self.provider_name} does not support cancellation"
)
async def validate_credentials(self) -> bool:
"""Validate that the stored credentials are correct."""
raise NotImplementedError(
f"{self.provider_name} does not support credential validation"
)The design decisions embedded in this class are worth examining:
- Three mandatory methods, three optional. Every provider must implement
initiate_payment,get_status, andhandle_webhook. Refunds, cancellations, and credential validation are optional because not every provider supports them.
- Credentials passed at construction. The provider instance receives decrypted credentials when it is created. This means the encryption layer is completely invisible to provider implementations.
InitPaymentResultis a dataclass, not a dict. Using a typed dataclass forces every provider to return the same structure. Thestatusfield is particularly important -- it tells the caller what to do next.
- The
dataparameter is a dict, not a model. This was a deliberate trade-off. Different providers need different fields (some need a phone number, others need an email, others need a redirect URL). A dict with documented keys was more flexible than a rigid model.
The InitPaymentResult Status Types
The status field in InitPaymentResult is the key to handling different payment flows uniformly:
| Status | Meaning | What the Caller Does |
|---|---|---|
pending | Payment is processing (USSD push sent, awaiting confirmation) | Poll for status updates |
redirect | Customer must be redirected to complete payment | Return redirect URL to frontend |
ussd_push | USSD prompt sent to customer's phone | Show "check your phone" message, poll for status |
failed | Payment initiation failed | Return error to merchant |
This abstraction is what allows the orchestrator to handle Stripe (redirect to checkout page), Hub2 (USSD push to phone), and PawaPay (redirect to hosted page) through the same code path.
How Stripe Implements the Interface
Stripe's adapter wraps the Checkout Sessions API:
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:
# Verify Stripe signature
import hmac, hashlib
signature = headers.get("stripe-signature", "")
# ... signature verification logic ...
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"}Key observations: Stripe returns status="redirect" with a redirect_url. The caller sends this URL to the merchant's frontend, which redirects the customer to Stripe's hosted checkout page. The webhook handler normalizes Stripe's event format into the standard 0fee.dev format.
How Hub2 Implements the Same Interface
Hub2 serves Francophone Africa with USSD push payments. The same interface, entirely different flow:
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 uses operator codes, not method names
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
)
async def get_status(self, provider_ref: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/payments/{provider_ref}",
headers={
"Authorization": f"Bearer {self.api_key}",
"Ocp-Apim-Subscription-Key":
self.subscription_key,
}
)
payment = response.json()
hub2_status_map = {
"succeeded": "completed",
"failed": "failed",
"pending": "pending",
"cancelled": "cancelled",
}
return {
"status": hub2_status_map.get(
payment.get("status"), "pending"
),
"provider_ref": provider_ref,
}
async def handle_webhook(
self, payload: dict, headers: dict
) -> dict:
payment = payload.get("data", {})
status_map = {
"succeeded": "completed",
"failed": "failed",
"cancelled": "cancelled",
}
return {
"transaction_id": payment.get("metadata", {})
.get("zerofee_txn"),
"status": status_map.get(
payment.get("status"), "pending"
),
"provider_ref": payment.get("id"),
}The contrast is stark. Where Stripe returns status="redirect", Hub2 returns status="ussd_push". Where Stripe provides a redirect_url, Hub2 provides instructions telling the customer to check their phone. But the calling code does not care:
python# In the payment route handler -- same code for every provider
result = await provider.initiate_payment(payment_data)
if result.status == "redirect":
# Send redirect URL to frontend
return {"payment_flow": "redirect", "url": result.redirect_url}
elif result.status == "ussd_push":
# Tell frontend to show "check your phone"
return {"payment_flow": "ussd_push", "message": result.instructions}
elif result.status == "pending":
# Payment is processing
return {"payment_flow": "pending"}
elif result.status == "failed":
# Return error
return {"error": "Payment initiation failed"}The Provider Registry
Provider adapters are not instantiated directly. Instead, a registry manages them:
pythonclass ProviderRegistry:
"""Registry for managing payment provider instances."""
_providers: dict[str, type[BasePayinProvider]] = {}
_instances: dict[str, BasePayinProvider] = {}
def register(
self, provider_id: str, provider_class: type[BasePayinProvider]
):
"""Register a provider class."""
self._providers[provider_id] = provider_class
def get_instance(
self, provider_id: str, credentials: dict,
app_id: str = ""
) -> BasePayinProvider:
"""
Get or create a provider instance.
Instances are cached by 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]:
"""List all registered providers."""
return [
{
"id": pid,
"name": cls.provider_name,
"countries": cls.supported_countries,
"methods": cls.supported_methods,
}
for pid, cls in self._providers.items()
]
# Global registry instance
provider_registry = ProviderRegistry()
# Register all providers at module load
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)The registry provides two important capabilities:
- Instance caching. Provider instances are cached by
provider_id:app_id, so credentials are loaded once and reused across requests. This avoids re-decrypting credentials on every payment.
- Dynamic management. New providers can be registered at runtime without restarting the application. When BUI and PaiementPro were added in Session 002, they were simply registered in the same registry.
Why Not Use a Plugin System?
Some payment orchestrators use plugin architectures with hot-loading, configuration files, and dynamic discovery. We chose a simpler approach:
- Providers are Python modules. Adding a new provider means creating a new directory with a
provider.pyfile and registering it in the registry. No configuration files, no plugin manifests, no discovery mechanisms. - Type safety is maintained. Because provider classes inherit from
BasePayinProvider, IDEs and type checkers can verify that every required method is implemented. - Testing is straightforward. The Test provider implements the same interface with deterministic results, making it trivial to test payment flows without mocking external APIs.
The adapter pattern is not novel. It is a textbook design pattern. But in the context of a payment orchestrator, where the complexity of provider integration is the core challenge, applying it consistently across every provider is what makes the system manageable. When 0fee.dev grew from 5 providers to 7 across Session 002, no existing code was modified -- only new adapter modules were added and registered.
The Pattern in Practice: Adding a New Provider
When BUI was added in Session 002, the process was:
- Create
backend/providers/bui/__init__.pyandbackend/providers/bui/provider.py. - Implement
BuiProvider(BasePayinProvider)with the three required methods. - Add one line to the registry:
provider_registry.register("bui", BuiProvider).
No changes to the payment route. No changes to the webhook handler. No changes to the routing engine. The new provider was immediately available to any app that configured BUI credentials.
This is the power of the adapter pattern in a payment orchestrator: the interface absorbs the complexity of each provider, and the rest of the system never has to know whether a payment went through Stripe's Checkout Sessions API or Hub2's USSD push mechanism. One interface, every payment system.
This article is part of the "How We Built 0fee.dev" series. 0fee.dev is a payment orchestrator covering 53+ providers across 200+ countries, built by Juste A. GNIMAVO and Claude from Abidjan with zero human engineers. Follow the series for the complete build story.