跳到主要内容

Python SDK

Pelago 官方 Python SDK。

安装​

pip install pelago-sdk

快速开始​

import os
from pelago import PelagoClient

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

# 创建支付
payment = pelago.payments.create(
    amount=99.99,
    currency='USD',
    cryptocurrency='USDC',
    network='stellar',
    merchant_wallet='GXXXXX...'
)

print(f'支付链接: {payment.url}')

配置​

pelago = PelagoClient(
    # 必填
    api_key='pk_live_xxxxx',
    api_secret='sk_live_xxxxx',
    
    # 可选
    environment='production',  # 'production' 或 'sandbox'
    timeout=30,               # 请求超时（秒）
    max_retries=3             # 自动重试失败请求
)

支付​

创建支付​

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]'
    }
)

查询支付​

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

列出支付​

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}')

退款​

# 创建退款
refund = pelago.refunds.create(
    payment_id='pay_abc123',
    amount=50.00,  # 部分退款
    reason='Customer request'
)

# 列出退款
refunds = pelago.refunds.list(payment_id='pay_abc123')

Webhook（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

错误处理​

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

try:
    payment = pelago.payments.create(...)
except AuthenticationError:
    print('请检查 API 密钥')
except InvalidRequestError as e:
    print(f'无效参数: {e.param}')
except RateLimitError:
    print('触发速率限制，请降低频率')
except PelagoError as e:
    print(f'错误: {e.message}')

Django 集成​

# 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})

异步支持​

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.url}')

asyncio.run(main())

下一步​