Python Examples

Ready-to-use Python code for common Pixlpay API operations.

Setup

Using requests

import requests
import os

class PixlpayClient:
    def __init__(self, subdomain: str, api_token: str):
        self.base_url = f"https://{subdomain}.pixlpay.net/api/external/v1"
        self.headers = {
            'Authorization': f'Bearer {api_token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
        }

    def get(self, endpoint: str, params: dict = None):
        response = requests.get(
            f"{self.base_url}{endpoint}",
            headers=self.headers,
            params=params
        )
        return response

    def post(self, endpoint: str, data: dict = None):
        response = requests.post(
            f"{self.base_url}{endpoint}",
            headers=self.headers,
            json=data
        )
        return response

Fetching Products

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

# Get all products
response = client.get('/products')
if response.status_code == 200:
    products = response.json()['data']
    for product in products:
        print(f"{product['name']} - ${product['price']}")

# Get single product
response = client.get('/products/1')
product = response.json()['data']

Fetching Orders

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

# Get pending orders
response = client.get('/orders', {'status': 'pending'})
orders = response.json()['data']

for order in orders:
    print(f"Order #{order['order_number']} - {order['customer']['email']}")

# Get orders by date
response = client.get('/orders', {'from': '2025-01-01', 'to': '2025-01-31'})

Fulfilling Orders

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

order_id = 1
response = client.post(f'/orders/{order_id}/fulfill', {
    'note': 'Delivered via API'
})

if response.status_code == 200:
    print('Order fulfilled successfully!')

Webhook Signature Verification

Pixlpay signs all webhooks with HMAC-SHA256. Always verify signatures before processing webhook payloads.

Signature Verification Function

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """
    Verify webhook signature using HMAC-SHA256.

    Args:
        payload: Raw request body as bytes (not parsed JSON)
        signature: Value from X-Webhook-Signature header
        secret: Your webhook secret

    Returns:
        True if signature is valid
    """
    # Compute HMAC-SHA256 signature
    computed = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    # Use timing-safe comparison to prevent timing attacks
    return hmac.compare_digest(computed, signature)

Flask Webhook Handler

from flask import Flask, request, abort, jsonify
import hmac
import hashlib
import os
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    computed = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@app.route('/webhooks/pixlpay', methods=['POST'])
def handle_webhook():
    # Get raw body BEFORE Flask parses it
    payload = request.data

    # Get headers
    signature = request.headers.get('X-Webhook-Signature', '')
    event_type = request.headers.get('X-Webhook-Event', '')
    delivery_id = request.headers.get('X-Webhook-ID', '')

    # Get secret from environment
    secret = os.environ['PIXLPAY_WEBHOOK_SECRET']

    # Verify signature FIRST
    if not verify_webhook_signature(payload, signature, secret):
        logger.warning(f'Invalid webhook signature for delivery: {delivery_id}')
        abort(401, 'Invalid signature')

    # Now safe to parse the payload
    event = request.json
    data = event.get('data', {})

    # Handle different events
    handlers = {
        'order.received': handle_order_received,
        'order.refunded': handle_order_refunded,
        'subscription.created': handle_subscription_created,
        'subscription.cancelled': handle_subscription_cancelled,
        'delivery.completed': handle_delivery_completed,
        'delivery.failed': handle_delivery_failed,
    }

    handler = handlers.get(event_type)
    if handler:
        handler(data)
    else:
        logger.info(f'Unhandled event type: {event_type}')

    # Respond quickly with 200
    return '', 200

def handle_order_received(order: dict):
    logger.info(f"Order received: {order['order_number']}")
    # Your delivery logic here

def handle_order_refunded(order: dict):
    logger.info(f"Order refunded: {order['order_number']}")
    # Revoke access, remove roles, etc.

def handle_subscription_created(subscription: dict):
    logger.info(f"Subscription created: {subscription['id']}")
    # Grant recurring access

def handle_subscription_cancelled(subscription: dict):
    logger.info(f"Subscription cancelled: {subscription['id']}")
    # Revoke recurring access

def handle_delivery_completed(delivery: dict):
    logger.info(f"Delivery completed: {delivery['id']}")

def handle_delivery_failed(delivery: dict):
    logger.info(f"Delivery failed: {delivery['id']}")
    # Alert admin, retry logic, etc.

if __name__ == '__main__':
    app.run(port=3000)

Webhook Verification with Idempotency

Prevent replay attacks by tracking processed webhooks:

from flask import Flask, request, abort
import hmac
import hashlib
import os
import redis
import logging

app = Flask(__name__)
logger = logging.getLogger(__name__)

# Connect to Redis
redis_client = redis.Redis.from_url(os.environ.get('REDIS_URL', 'redis://localhost:6379'))

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    computed = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@app.route('/webhooks/pixlpay', methods=['POST'])
def handle_webhook():
    payload = request.data
    signature = request.headers.get('X-Webhook-Signature', '')
    delivery_id = request.headers.get('X-Webhook-ID', '')
    secret = os.environ['PIXLPAY_WEBHOOK_SECRET']

    # Verify signature
    if not verify_webhook_signature(payload, signature, secret):
        logger.warning(f'Invalid webhook signature for delivery: {delivery_id}')
        abort(401, 'Invalid signature')

    # Check for replay attack (idempotency)
    cache_key = f'pixlpay_webhook:{delivery_id}'
    if redis_client.exists(cache_key):
        # Already processed - return success but don't reprocess
        return 'Already processed', 200

    # Process the webhook
    event = request.json
    process_event(event)

    # Mark as processed (expire after 24 hours)
    redis_client.setex(cache_key, 86400, '1')

    return '', 200

def process_event(event: dict):
    event_type = event.get('event_type', '')
    data = event.get('data', {})
    logger.info(f'Processing event: {event_type}')
    # Your processing logic here

if __name__ == '__main__':
    app.run(port=3000)

Django Webhook Handler

import hmac
import hashlib
import json
import os
import logging
from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.core.cache import cache

logger = logging.getLogger(__name__)

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    computed = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@csrf_exempt
@require_POST
def pixlpay_webhook(request):
    # Get raw body
    payload = request.body

    # Get headers
    signature = request.headers.get('X-Webhook-Signature', '')
    event_type = request.headers.get('X-Webhook-Event', '')
    delivery_id = request.headers.get('X-Webhook-ID', '')

    # Get secret from settings
    secret = os.environ['PIXLPAY_WEBHOOK_SECRET']

    # Verify signature
    if not verify_webhook_signature(payload, signature, secret):
        logger.warning(f'Invalid webhook signature for delivery: {delivery_id}')
        return HttpResponseForbidden('Invalid signature')

    # Check for replay attack (idempotency)
    cache_key = f'pixlpay_webhook:{delivery_id}'
    if cache.get(cache_key):
        return HttpResponse('Already processed', status=200)

    # Parse and process
    try:
        event = json.loads(payload)
        data = event.get('data', {})

        # Handle based on event type
        if event_type == 'order.received':
            handle_order_received(data)
        elif event_type == 'order.refunded':
            handle_order_refunded(data)
        elif event_type == 'subscription.created':
            handle_subscription_created(data)
        elif event_type == 'subscription.cancelled':
            handle_subscription_cancelled(data)
        else:
            logger.info(f'Unhandled event type: {event_type}')

        # Mark as processed (cache for 24 hours)
        cache.set(cache_key, True, timeout=86400)

    except json.JSONDecodeError:
        return HttpResponse('Invalid JSON', status=400)

    return HttpResponse('OK', status=200)

def handle_order_received(order: dict):
    logger.info(f"Order received: {order['order_number']}")
    # Your delivery logic here

def handle_order_refunded(order: dict):
    logger.info(f"Order refunded: {order['order_number']}")

def handle_subscription_created(subscription: dict):
    logger.info(f"Subscription created: {subscription['id']}")

def handle_subscription_cancelled(subscription: dict):
    logger.info(f"Subscription cancelled: {subscription['id']}")

FastAPI Webhook Handler

from fastapi import FastAPI, Request, HTTPException, Header
import hmac
import hashlib
import os
import logging
from typing import Optional

app = FastAPI()
logger = logging.getLogger(__name__)

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    computed = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@app.post('/webhooks/pixlpay')
async def handle_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None),
    x_webhook_event: Optional[str] = Header(None),
    x_webhook_id: Optional[str] = Header(None),
):
    # Get raw body
    payload = await request.body()

    # Get secret from environment
    secret = os.environ['PIXLPAY_WEBHOOK_SECRET']

    # Verify signature
    if not x_webhook_signature or not verify_webhook_signature(payload, x_webhook_signature, secret):
        logger.warning(f'Invalid webhook signature for delivery: {x_webhook_id}')
        raise HTTPException(status_code=401, detail='Invalid signature')

    # Parse and process
    event = await request.json()
    data = event.get('data', {})

    # Handle based on event type
    handlers = {
        'order.received': handle_order_received,
        'order.refunded': handle_order_refunded,
        'subscription.created': handle_subscription_created,
        'subscription.cancelled': handle_subscription_cancelled,
    }

    handler = handlers.get(x_webhook_event)
    if handler:
        await handler(data)
    else:
        logger.info(f'Unhandled event type: {x_webhook_event}')

    return {'status': 'ok'}

async def handle_order_received(order: dict):
    logger.info(f"Order received: {order['order_number']}")

async def handle_order_refunded(order: dict):
    logger.info(f"Order refunded: {order['order_number']}")

async def handle_subscription_created(subscription: dict):
    logger.info(f"Subscription created: {subscription['id']}")

async def handle_subscription_cancelled(subscription: dict):
    logger.info(f"Subscription cancelled: {subscription['id']}")

Creating Webhooks

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

response = client.post('/webhooks', {
    'url': 'https://yoursite.com/webhooks/pixlpay',
    'events': ['order.received', 'order.refunded', 'subscription.cancelled']
})

if response.status_code == 201:
    webhook = response.json()['data']
    print(f"Webhook created! Secret: {webhook['secret']}")

    # IMPORTANT: Store this secret securely - you'll need it for verification
    # Save to .env or secure config

Analytics

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

response = client.get('/analytics/revenue', {
    'from': '2025-01-01',
    'to': '2025-01-31'
})

if response.status_code == 200:
    data = response.json()['data']
    print(f"Revenue: ${data['total_revenue']}")
    print(f"Orders: {data['total_orders']}")
    print(f"AOV: ${data['average_order_value']}")

Error Handling

client = PixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])

response = client.get('/products/99999')

if response.status_code != 200:
    error = response.json()
    print(f"Error: {error.get('message', 'Unknown error')}")

    error_handlers = {
        401: lambda: print('Check your API token'),
        404: lambda: print('Resource not found'),
        429: lambda: print('Rate limited - slow down!'),
    }

    handler = error_handlers.get(response.status_code)
    if handler:
        handler()

Async with aiohttp

import aiohttp
import asyncio

class AsyncPixlpayClient:
    def __init__(self, subdomain: str, api_token: str):
        self.base_url = f"https://{subdomain}.pixlpay.net/api/external/v1"
        self.headers = {
            'Authorization': f'Bearer {api_token}',
            'Accept': 'application/json',
        }

    async def get(self, endpoint: str):
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}{endpoint}",
                headers=self.headers
            ) as response:
                return await response.json()

# Usage
async def main():
    client = AsyncPixlpayClient('yourstore', os.environ['PIXLPAY_TOKEN'])
    products = await client.get('/products')
    print(products)

asyncio.run(main())