A scalable, API-first monorepo (backend + frontend) for orchestrating multi-gateway payments across Paystack, Flutterwave, and Moniepoint.
This project provides a unified interface for payment initialization, verification, OTP handling, and webhook management designed to make multi-gateway operations seamless, reliable, and developer-friendly.
- Unified API Interface: Interact with multiple gateways (Paystack, Moniepoint) through a consistent set of endpoints.
- Gateway Abstraction Layer: Easily extend to new providers by subclassing the
PaymentsService. - Secure Authentication: Email + password authentication with JWT-protected routes.
- Transactions Management: Centralized transaction model with unique references and metadata tracking.
- Webhook Resilience: Queue-based webhook retry system using Celery + Redis.
- OTP Handling: Endpoints for submitting OTPs (for Paystack no-redirect flows).
- Metrics & Logging: Per-gateway metrics for success/failure rates and response times.
- API Documentation: OpenAPI 3.0 via Swagger UI (
/apidocs).
| Layer | Tech |
|---|---|
| Framework | Flask |
| Auth | JWT (PyJWT) |
| ORM | SQLAlchemy |
| Database | SQLite (dev), PostgreSQL/MySQL (prod) |
| API Docs | Flasgger (Swagger UI) |
| Environment | python-dotenv |
Create a .env file based on .env.example:
FLASK_ENV=development
SECRET_KEY=your_secret_key
DATABASE_URL=sqlite:///app.db
# Paystack
PAYSTACK_SECRET_KEY=sk_test_xxxxxxxxxx
PAYSTACK_PUBLIC_KEY=pk_test_xxxxxxxxxx
PAYSTACK_BASE_URL=https://api.paystack.co
# Moniepoint (if applicable)
MONIEPOINT_SECRET_KEY=sk_test_xxxxxxxxxx
MONIEPOINT_BASE_URL=https://sandbox-api.moniepoint.com
# Celery / Redis
REDIS_URL=redis://localhost:6379/0git clone https://github.com/duowork/payment-orchestration-backend.git
cd payment-orchestration-backend
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
flask run
- Create a new file in payments/ (e.g., flutterwave.py).
- Subclass
PaymentsServiceand implement:initialize_charge()verify_payment()handle_webhook()
class FlutterwaveService(PaymentsService):
def initialize_charge(self, data): ...
def verify_payment(self, reference): ...
def handle_webhook(self, payload): ...