Skip to main content

Python SDK

The official Python SDK for Pelago.

Installation​

pip install pelago-sdk

Quick Start​

import os
from pelago import PelagoClient

pelago = PelagoClient(
    api_key=os.environ['PELAGO_API_KEY'],
    api_secret=os.environ['PELAGO_API_SECRET'],
    environment='production'
)

# Create a payment
payment = pelago.payments.create(
    amount=99.99,
    currency='USD',
    cryptocurrency='USDC',
    network='stellar',
    merchant_wallet='GXXXXX...'
)

print(f'Payment URL: {payment.url}')

Configuration​

pelago = PelagoClient(
    # Required
    api_key='pk_live_xxxxx',
    api_secret='sk_live_xxxxx',
    
    # Optional
    environment='production',  # 'production' or 'sandbox'
    timeout=30,               # Request timeout in seconds
    max_retries=3             # Auto-retry failed requests
)

Payments​

Create Payment​

payment = pelago.payments.create(
    amount=100.00,
    currency='USD',
    cryptocurrency='USDC',
    network='stellar',
    merchant_wallet='GXXXXX...',
    redirect_url='https://mystore.com/success',
    webhook_url='https://mystore.com/api/webhook',
    expires_in=1800,
    metadata={
        'order_id': 'ORD-12345',
        'customer_email': '[email protected]'
    }
)

Retrieve Payment​

payment = pelago.payments.retrieve('pay_abc123')
print(f'Status: {payment.status}')

List Payments​

payments = pelago.payments.list(
    status='completed',
    limit=20,
    created_gte='2025-01-01T00:00:00Z'
)

for payment in payments.data:
    print(f'{payment.id}: {payment.amount}')

Refunds​

# Create refund
refund = pelago.refunds.create(
    payment_id='pay_abc123',
    amount=50.00,  # Partial refund
    reason='Customer request'
)

# List refunds
refunds = pelago.refunds.list(payment_id='pay_abc123')

Webhooks (Flask)​

from flask import Flask, request, jsonify
from pelago import PelagoClient
from pelago.errors import WebhookSignatureError

app = Flask(__name__)

pelago = PelagoClient(
    api_key=os.environ['PELAGO_API_KEY'],
    api_secret=os.environ['PELAGO_API_SECRET']
)

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_data(as_text=True)
    signature = request.headers.get('X-Pelago-Signature')
    timestamp = request.headers.get('X-Pelago-Timestamp')
    
    try:
        event = pelago.webhooks.construct_event(
            payload,
            signature,
            timestamp,
            os.environ['WEBHOOK_SECRET']
        )
    except WebhookSignatureError:
        return jsonify({'error': 'Invalid signature'}), 401
    
    if event['type'] == 'payment.completed':
        order_id = event['data']['metadata']['order_id']
        fulfill_order(order_id)
    
    return jsonify({'received': True}), 200

Error Handling​

from pelago.errors import (
    PelagoError,
    AuthenticationError,
    InvalidRequestError,
    RateLimitError
)

try:
    payment = pelago.payments.create(...)
except AuthenticationError:
    print('Check your API keys')
except InvalidRequestError as e:
    print(f'Invalid param: {e.param}')
except RateLimitError:
    print('Rate limited, slow down')
except PelagoError as e:
    print(f'Error: {e.message}')

Django Integration​

# views.py
import json
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

@csrf_exempt
@require_POST
def create_payment(request):
    data = json.loads(request.body)
    
    payment = pelago.payments.create(
        amount=data['amount'],
        currency='USD',
        cryptocurrency='USDC',
        network='stellar',
        merchant_wallet=settings.PELAGO_WALLET,
        metadata={'order_id': data['order_id']}
    )
    
    return JsonResponse({'payment_url': payment.url})

@csrf_exempt
@require_POST
def webhook(request):
    event = pelago.webhooks.construct_event(
        request.body,
        request.headers.get('X-Pelago-Signature'),
        request.headers.get('X-Pelago-Timestamp'),
        settings.WEBHOOK_SECRET
    )
    
    if event['type'] == 'payment.completed':
        Order.objects.filter(
            id=event['data']['metadata']['order_id']
        ).update(status='paid')
    
    return JsonResponse({'received': True})

Async Support​

import asyncio
from pelago import AsyncPelagoClient

pelago = AsyncPelagoClient(
    api_key=os.environ['PELAGO_API_KEY'],
    api_secret=os.environ['PELAGO_API_SECRET']
)

async def main():
    payment = await pelago.payments.create(
        amount=100,
        currency='USD',
        cryptocurrency='USDC',
        network='stellar',
        merchant_wallet='GXXXXX...'
    )
    print(f'Payment: {payment.url}')

asyncio.run(main())

Next Steps​