Best Practices
This guide provides recommendations for integrating the JPay Africa API securely and efficiently.
Authentication Best Practices
1. Secure Token Storage
Store tokens securely based on your environment:
Server-Side (Backend)
import os
from dotenv import load_dotenv
load_dotenv()
# Load from environment variables
APP_KEY = os.getenv('JPAY_APP_KEY')
APP_SECRET = os.getenv('JPAY_APP_SECRET')
# Never hardcode credentials
NEVER_DO_THIS = {
'app_key': 'YOUR_ACTUAL_KEY', # DO NOT DO THIS
'app_secret': 'YOUR_ACTUAL_SECRET' # DO NOT DO THIS
}
# .env file (add to .gitignore)
# JPAY_APP_KEY=your_key_here
# JPAY_APP_SECRET=your_secret_here
Web Application (Frontend)
// Bad: Store in localStorage (vulnerable to XSS)
localStorage.setItem('access_token', token); // DO NOT DO THIS
// Good: Use httpOnly cookies (server sets, client cannot access)
// Server should set:
// Set-Cookie: access_token=...; HttpOnly; Secure; SameSite=Strict
// Client will auto-include in requests
2. Token Refresh Strategy
Refresh tokens proactively before expiration:
from datetime import datetime, timedelta
import jwt
class TokenManager:
def __init__(self, access_token, refresh_token):
self.access_token = access_token
self.refresh_token = refresh_token
self.token_expiry = self._decode_expiry(access_token)
def _decode_expiry(self, token):
"""Decode JWT to get expiry time"""
try:
decoded = jwt.decode(token, options={"verify_signature": False})
return datetime.fromtimestamp(decoded['exp'])
except:
return datetime.now()
def should_refresh(self):
"""Check if token should be refreshed (within 5 minutes of expiry)"""
time_until_expiry = self.token_expiry - datetime.now()
return time_until_expiry < timedelta(minutes=5)
def get_valid_token(self):
"""Get valid token, refreshing if necessary"""
if self.should_refresh():
self.refresh()
return self.access_token
def refresh(self):
"""Refresh the access token"""
response = requests.post(
'https://sandbox.api.jpay.africa/api/v1/auth/refresh',
json={'refresh': self.refresh_token}
)
if response.status_code == 200:
data = response.json()
self.access_token = data['access']
self.refresh_token = data['refresh']
self.token_expiry = self._decode_expiry(self.access_token)
3. Credential Rotation
Rotate credentials regularly:
1. Generate new app_key and app_secret in dashboard
2. Update environment variables with new credentials
3. Test with new credentials before discarding old ones
4. Disable old credentials in dashboard
5. Schedule quarterly credential rotation
Data Validation Best Practices
1. Phone Number Validation
Always validate phone numbers in the correct format:
import phonenumbers
from typing import Optional
def validate_phone_number(phone: str, country_code: str = "KE") -> Optional[str]:
"""Validate and format phone number to E.164 format"""
try:
parsed = phonenumbers.parse(phone, country_code)
if not phonenumbers.is_valid_number(parsed):
return None
return phonenumbers.format_number(
parsed,
phonenumbers.PhoneNumberFormat.E164
)
except phonenumbers.NumberParseException:
return None
# Usage
phone = validate_phone_number("0712345678") # "0712345678"
if phone:
print(f"Valid: {phone}") # Valid: +254712345678
else:
print("Invalid phone number")
2. Amount Validation
Validate amounts before submission:
from decimal import Decimal
from typing import Tuple
def validate_amount(amount: str) -> Tuple[bool, Optional[Decimal], str]:
"""Validate payment amount"""
try:
decimal_amount = Decimal(str(amount))
# Check range
if decimal_amount < Decimal('1.00'):
return False, None, "Amount must be at least 1.00"
if decimal_amount > Decimal('999999.99'):
return False, None, "Amount cannot exceed 999999.99"
# Check decimal places
if decimal_amount.as_tuple().exponent < -2:
return False, None, "Amount must have maximum 2 decimal places"
# Format to exactly 2 decimal places
formatted = f"{decimal_amount:.2f}"
return True, Decimal(formatted), "Valid"
except:
return False, None, "Invalid amount format"
# Usage
valid, amount, message = validate_amount("1000.50")
if valid:
print(f"Valid amount: {amount}") # Valid amount: 1000.50
else:
print(f"Error: {message}")
3. Reference Number Validation
Ensure unique, valid reference numbers:
import uuid
from datetime import datetime
import re
def generate_reference_number(prefix: str = "TXNREF") -> str:
"""Generate a unique reference number"""
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
unique_id = str(uuid.uuid4())[:6].upper()
return f"{prefix}-{timestamp}-{unique_id}"
def validate_reference_number(ref_no: str) -> bool:
"""Validate reference number format"""
# Max 50 characters
if len(ref_no) > 50:
return False
# Alphanumeric, hyphens, underscores only
if not re.match(r'^[a-zA-Z0-9\-_]+$', ref_no):
return False
return True
# Usage
ref_no = generate_reference_number("COLL")
print(ref_no) # COLL-20241204143045-ABC123
# Validate
if validate_reference_number(ref_no):
print("Reference number is valid")
Request/Response Handling
1. Implement Retry Logic
Use exponential backoff for transient failures:
import time
import requests
from typing import Optional
def api_request_with_retry(
method: str,
url: str,
headers: dict,
json_data: Optional[dict] = None,
max_retries: int = 3,
base_delay: int = 1
) -> Optional[requests.Response]:
"""Make API request with exponential backoff retry"""
for attempt in range(max_retries):
try:
if method == 'GET':
response = requests.get(url, headers=headers, timeout=10)
elif method == 'POST':
response = requests.post(url, headers=headers, json=json_data, timeout=10)
else:
raise ValueError(f"Unsupported method: {method}")
# Don't retry on client validation errors
if 400 <= response.status_code < 500 and response.status_code != 429:
return response
# Retry on rate limit and server errors
if response.status_code == 429 or response.status_code >= 500:
if attempt < max_retries - 1:
wait_time = base_delay * (2 ** attempt)
print(f"Attempt {attempt + 1} failed with {response.status_code}. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
return response
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait_time = base_delay * (2 ** attempt)
print(f"Request timeout. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
except requests.exceptions.ConnectionError:
if attempt < max_retries - 1:
wait_time = base_delay * (2 ** attempt)
print(f"Connection error. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
return None
2. Implement Idempotency
Use unique reference numbers to ensure idempotent operations:
def create_collection_idempotent(
token_manager,
customer_phone: str,
amount: str,
order_id: str
) -> dict:
"""Create collection with idempotency guarantee"""
# Use order_id as reference to ensure idempotency
headers = token_manager.get_auth_headers()
payload = {
'payfrom': customer_phone,
'amount': amount,
'ref_no': order_id, # Unique identifier
'account_number': '1001'
}
# Multiple requests with same ref_no should be idempotent
response = requests.post(
'https://sandbox.api.jpay.africa/api/v1/payments/collections/checkouts/initiate',
headers=headers,
json=payload
)
return response.json() if response.status_code == 200 else None
# Usage - Safe to retry without creating duplicates
for attempt in range(3):
result = create_collection_idempotent(
token_manager,
'+254712345678',
'1000.00',
'ORDER-2024-001' # Same ref_no on retries
)
if result:
break
3. Handle Webhooks Securely
Validate webhook requests from JPay:
import hashlib
import hmac
import json
from flask import request
WEBHOOK_SECRET = os.getenv('JPAY_WEBHOOK_SECRET')
@app.route('/webhook/collections', methods=['POST'])
def handle_collection_webhook():
"""Handle collection completed webhook"""
# Verify signature
signature = request.headers.get('X-JPay-Signature')
body = request.get_data()
# Calculate expected signature
expected_signature = hmac.new(
WEBHOOK_SECRET.encode(),
body,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
return {'error': 'Invalid signature'}, 401
# Process webhook
data = request.json
event = data['event']
transaction_id = data['transaction_id']
status = data['status']
# Update database
if status == 'completed':
update_order_status(data['ref_no'], 'PAID')
deliver_product()
elif status == 'failed':
update_order_status(data['ref_no'], 'FAILED')
notify_customer()
return {'success': True}, 200
Error Handling Best Practices
1. Implement Comprehensive Logging
Log all API interactions for debugging:
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
def log_api_request(method, url, payload, response_code, response_body):
"""Log API request and response"""
logger.info(
f"API Request",
extra={
'method': method,
'url': url,
'request_payload': payload,
'response_code': response_code,
'response_body': response_body,
'timestamp': datetime.now().isoformat()
}
)
def log_error(error_type, error_message, context):
"""Log errors with context"""
logger.error(
f"Error: {error_type}",
extra={
'error_message': error_message,
'context': context,
'timestamp': datetime.now().isoformat()
}
)
# Usage
try:
response = make_api_request()
log_api_request('POST', endpoint, payload, response.status_code, response.json())
except Exception as e:
log_error('API_ERROR', str(e), {'endpoint': endpoint, 'payload': payload})
2. Handle Different Error Types
def handle_api_error(response, request_context):
"""Handle different types of API errors"""
error_data = response.json()
detail = error_data.get('detail', 'Unknown error')
if response.status_code == 400:
# Client error - validation issue
log_error('VALIDATION_ERROR', detail, request_context)
return {
'status': 'error',
'message': 'Please check your input and try again',
'detail': detail
}
elif response.status_code == 401:
# Authentication error
log_error('AUTH_ERROR', detail, request_context)
return {
'status': 'error',
'message': 'Session expired. Please log in again',
'redirect': '/login'
}
elif response.status_code == 403:
# Authorization error
log_error('PERMISSION_ERROR', detail, request_context)
return {
'status': 'error',
'message': 'You do not have permission to perform this action'
}
elif response.status_code == 429:
# Rate limit
log_error('RATE_LIMIT', detail, request_context)
return {
'status': 'error',
'message': 'Too many requests. Please wait and try again',
'retry_after': 60
}
elif response.status_code >= 500:
# Server error - retry
log_error('SERVER_ERROR', detail, request_context)
return {
'status': 'error',
'message': 'Server error. Please try again later',
'retry': True
}
Rate Limiting Best Practices
1. Implement Request Throttling
import time
from functools import wraps
class RateLimiter:
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
self.min_interval = 60 / requests_per_minute
self.last_request_time = 0
def wait_if_needed(self):
"""Wait if necessary to respect rate limit"""
time_since_last = time.time() - self.last_request_time
if time_since_last < self.min_interval:
time.sleep(self.min_interval - time_since_last)
self.last_request_time = time.time()
# Usage
rate_limiter = RateLimiter(requests_per_minute=50)
for item in items:
rate_limiter.wait_if_needed()
make_api_request(item)
2. Monitor Rate Limit Headers
def make_request_and_monitor(endpoint, headers):
"""Make request and monitor rate limit"""
response = requests.get(endpoint, headers=headers)
remaining = int(response.headers.get('X-RateLimit-Remaining', 0))
limit = int(response.headers.get('X-RateLimit-Limit', 60))
# Warn if approaching limit
if remaining < 5:
logger.warning(f"Approaching rate limit: {remaining}/{limit} requests remaining")
# Pause if nearly limited
if remaining < 2:
logger.warning("Rate limit nearly exceeded. Pausing requests.")
time.sleep(5)
return response
Performance Optimization
1. Batch Operations
def batch_create_collections(collections: list, batch_size: int = 10):
"""Create collections in batches to manage load"""
for i in range(0, len(collections), batch_size):
batch = collections[i:i + batch_size]
results = []
for collection in batch:
result = create_collection(collection)
results.append(result)
# Process results before next batch
yield results
# Add delay between batches
if i + batch_size < len(collections):
time.sleep(1)
2. Use Connection Pooling
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_pooling():
"""Create requests session with connection pooling"""
session = requests.Session()
# Configure connection pooling
adapter = HTTPAdapter(
pool_connections=10,
pool_maxsize=10,
max_retries=Retry(total=3, backoff_factor=0.5)
)
session.mount('https://', adapter)
session.mount('http://', adapter)
return session
# Usage
session = create_session_with_pooling()
response = session.get(endpoint)
3. Implement Caching
from functools import lru_cache
from datetime import datetime, timedelta
class CachedAPIClient:
def __init__(self):
self.cache = {}
self.cache_ttl = timedelta(minutes=5)
def get_with_cache(self, key: str, fetch_func, *args):
"""Get data from cache or fetch if expired"""
if key in self.cache:
data, timestamp = self.cache[key]
if datetime.now() - timestamp < self.cache_ttl:
return data
# Fetch fresh data
data = fetch_func(*args)
self.cache[key] = (data, datetime.now())
return data
# Usage
client = CachedAPIClient()
collections = client.get_with_cache(
'collections_list',
list_collections,
token
)
Security Best Practices
1. Configure IP Whitelisting
Restrict API access to specific IP addresses for enhanced security:
# In your merchant dashboard, configure allowed IPs:
# App Settings → Allowed IPs → Add IP addresses
# Example: Production server IPs
allowed_ips = [
"203.0.113.42", # Production server 1
"203.0.113.43", # Production server 2
"198.51.100.0/24" # Office network (CIDR notation)
]
Benefits:
- Prevents unauthorized API access even if credentials are compromised
- Protects against token theft and replay attacks
- Required for PCI-DSS and financial compliance
Best Practices:
- Use IP whitelisting for all production apps
- Regularly audit and update IP whitelist
- Document all whitelisted IPs and their purposes
- Use static IPs or reserved IP ranges for production servers
- Leave IP restrictions empty only for development/testing
tip
Finding Your Server IP:
# On your server, run:
curl https://api.ipify.org
# Or check request IP in application:
import requests
response = requests.get('https://api.ipify.org?format=json')
print(response.json()['ip'])
2. Use HTTPS Only
# Always use HTTPS, never HTTP
API_URL = "https://sandbox.api.jpay.africa/api/v1" # Good
# API_URL = "http://api.jpay.africa/api/v1" # Bad
3. Validate SSL Certificates
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.ssl_ import create_urllib3_context
class SSLAdapter(HTTPAdapter):
def init_poolmanager(self, *args, **kwargs):
context = create_urllib3_context()
kwargs['ssl_context'] = context
return super().init_poolmanager(*args, **kwargs)
session = requests.Session()
session.mount('https://', SSLAdapter())
4. Sanitize Sensitive Data in Logs
import logging
import re
class SensitiveDataFilter(logging.Filter):
def filter(self, record):
# Remove tokens from logs
record.msg = re.sub(
r'Bearer\s+[a-zA-Z0-9\-_.]+',
'Bearer [REDACTED]',
str(record.msg)
)
# Remove app_secret
record.msg = re.sub(
r'"app_secret"\s*:\s*"[^"]*"',
'"app_secret": "[REDACTED]"',
str(record.msg)
)
return True
# Apply filter to logger
logger.addFilter(SensitiveDataFilter())
Testing Best Practices
1. Use Test Credentials
# Testing configuration
TEST_CONFIG = {
'app_code': '12345',
'test_phone': '+254712345678',
'test_amount': '100.00',
'sandbox_mode': True
}
# Production configuration
PROD_CONFIG = {
'app_code': os.getenv('PROD_APP_CODE'),
'sandbox_mode': False
}
2. Mock API Responses for Testing
import unittest
from unittest.mock import patch, MagicMock
class TestJPayIntegration(unittest.TestCase):
@patch('requests.post')
def test_create_collection(self, mock_post):
# Mock successful response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
'ref_no': 'TEST-001',
'status': 'pending'
}
mock_post.return_value = mock_response
# Test your code
result = create_collection_function()
assert result['status'] == 'pending'
Monitoring and Alerting
1. Monitor Key Metrics
def track_transaction_metrics(response, transaction_type):
"""Track transaction success/failure rates"""
if response.status_code == 200:
metrics['transactions_successful'] += 1
else:
metrics['transactions_failed'] += 1
# Alert if failure rate exceeds threshold
failure_rate = metrics['transactions_failed'] / total_transactions
if failure_rate > 0.05: # 5% threshold
send_alert(f"Transaction failure rate: {failure_rate * 100}%")
Compliance
1. Data Privacy
- Do not log sensitive customer data
- Encrypt sensitive data in transit (HTTPS) and at rest
- Follow GDPR/CCPA regulations for customer data
- Securely delete data after retention period
2. PCI Compliance
- JPay Africa is PCI DSS Level 1 certified
- Do not store payment card data
- Use the API instead of direct card processing
- Validate SSL certificates
Related Resources
- Error Handling - Comprehensive error handling
- Authentication Overview - Authentication methods
- API Endpoints - Endpoint documentation