Setup
pip install requests
# settings.py
PAYCHAINLY_API_KEY = env('PAYCHAINLY_API_KEY')
PAYCHAINLY_WEBHOOK_SECRET = env('PAYCHAINLY_WEBHOOK_SECRET')
API Client
import requests, hmac, hashlib
from django.conf import settings
class PaychainlyClient:
BASE = 'https://paychainly.com'
def create_session(self, user_id: str, amount: float) -> dict:
r = requests.post(
f'{self.BASE}/api/v1/addresses/start-session',
json={'userId': user_id, 'amount': amount},
headers={'x-api-key': settings.PAYCHAINLY_API_KEY},
timeout=10,
)
r.raise_for_status()
return r.json()
Webhook View
import json, hmac, hashlib
from django.views.decorators.csrf import csrf_exempt
from django.http import JsonResponse, HttpResponse
@csrf_exempt
def paychainly_webhook(request):
if request.method != 'POST':
return HttpResponse(status=405)
payload = json.loads(request.body)
signature = request.headers.get('X-Signature', '')
fields = [payload['event'], payload['txHash'], payload['fromAddress'],
payload['toAddress'], payload['amount'], str(payload['blockNumber']),
str(payload['timestamp']), payload['userId']]
msg = '|'.join(fields)
expected = hmac.new(
settings.PAYCHAINLY_WEBHOOK_SECRET.encode(),
msg.encode(), hashlib.sha256
).hexdigest()
if not hmac.compare_digest(expected, signature):
return HttpResponse(status=401)
if payload['event'] == 'deposit_detected':
from .tasks import process_payment
process_payment.delay(payload)
return JsonResponse({'ok': True})
Celery Task
@shared_task(bind=True, max_retries=3)
def process_payment(self, payload):
Payment.objects.get_or_create(
tx_hash=payload['txHash'],
defaults={'order_id': payload['userId'],
'amount': payload['amount'], 'status': 'confirmed'}
)