KEY INSIGHT
Paystack provides Nigerian payment infrastructure with Naira support, recurring billing via authorization codes, and webhook-driven payment confirmation.
Paystack integration enables Nigerian customers to pay with local payment methods—cards, bank transfers, USSD. The integration requires handling payment initialization, webhook verification, and subscription lifecycle management.
Payment flow: initialize transaction → redirect customer to Paystack → receive webhook confirmation → update subscription status.
```python
import requests
from typing import Optional
class PaystackService:
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.base_url = "https://api.paystack.co"
self.headers = {
"Authorization": f"Bearer {secret_key}",
"Content-Type": "application/json"
}
def initialize_transaction(
self,
email: str,
amount_kobo: int,
reference: str,
callback_url: str,
metadata: dict
) -> dict:
"""Initialize a Paystack payment."""
payload = {
"email": email,
"amount": amount_kobo,
"reference": reference,
"callback_url": callback_url,
"metadata": metadata
}
response = requests.post(
f"{self.base_url}/transaction/initialize",
json=payload,
headers=self.headers
)
if response.status_code != 200:
raise PaystackError(f"Initialization failed: {response.text}")
data = response.json()
if not data["status"]:
raise PaystackError(f"Paystack error: {data['message']}")
return data["data"]
def verify_transaction(self, reference: str) -> dict:
"""Verify a transaction by reference."""
response = requests.get(
f"{self.base_url}/transaction/verify/{reference}",
headers=self.headers
)
if response.status_code != 200:
raise PaystackError(f"Verification failed: {response.text}")
data = response.json()
if not data["status"]:
raise PaystackError(f"Verification failed: {data['message']}")
return data["data"]
def charge_authorization(
self,
authorization_code: str,
email: str,
amount_kobo: int,
reference: str,
metadata: dict
) -> dict:
"""Charge a previously authorized payment method (recurring)."""
payload = {
"authorization_code": authorization_code,
"email": email,
"amount": amount_kobo,
"reference": reference,
"metadata": metadata
}
response = requests.post(
f"{self.base_url}/transaction/charge_authorization",
json=payload,
headers=self.headers
)
data = response.json()
return data["data"] if data["status"] else {"status": "failed", "message": data.get("message")}
class PaystackWebhookHandler:
def __init__(self, db: Session, paystack: PaystackService):
self.db = db
self.paystack = paystack
def handle_event(self, event: dict) -> None:
"""Process Paystack webhook events."""
event_type = event.get("event")
payload = event.get("data", {})
# Verify payload signature in production
# if not self._verify_signature(payload, headers):
# raise ValueError("Invalid webhook signature")
if event_type == "charge.success":
self._handle_successful_charge(payload)
elif event_type == "subscription.create":
self._handle_subscription_created(payload)
elif event_type == "subscription.disable":
self._handle_subscription_disabled(payload)
elif event_type == "subscription.not_renew":
self._handle_subscription_expiring(payload)
def _handle_successful_charge(self, data: dict) -> None:
"""Handle successful payment."""
reference = data["reference"]
amount = data["amount"] # Already in kobo
metadata = data.get("metadata", {})
organization_id = metadata.get("organization_id")
description = metadata.get("description", "Payment")
# Record payment
payment = Payment(
id=str(uuid.uuid4()),
organization_id=organization_id,
amount_kobo=amount,
reference=reference,
status="completed",
description=description,
paid_at=datetime.fromisoformat(data["paid_at"])
)
self.db.add(payment)
self._update_subscription_if_needed(organization_id, amount)
self.db.commit()
def _update_subscription_if_needed(self, organization_id: str, amount: int) -> None:
"""Extend subscription period based on payment amount."""
# Implementation depends on pricing structure
pass
```
Webhook security is critical. Paystack webhooks can be replayed or spoofed. Always verify the signature and check for duplicate event processing.
```python
# Webhook signature verification
import hmac
import hashlib
class PaystackWebhookVerifier:
def __init__(self, secret_key: str):
self.secret_key = secret_key
def verify(self, payload: bytes, headers: dict) -> bool:
"""Verify Paystack webhook signature."""
signature = headers.get("x-paystack-signature")
if not signature:
return False
expected = hmac.new(
self.secret_key.encode(),
payload,
hashlib.sha512
).hexdigest()
return hmac.compare_digest(signature, expected)
```