Python Examples

Complete Python examples for integrating with the Pixlpay API.

API Client Class

import os
import hmac
import hashlib
import requests
from typing import Dict, List, Optional, Any
from dataclasses import dataclass

@dataclass
class PixlpayError(Exception):
    message: str
    code: str
    status: int
    errors: Optional[Dict] = None

class PixlpayClient:
    def __init__(self, store_url: str, token: str):
        self.base_url = f"{store_url.rstrip('/')}/api/external/v1"
        self.token = token
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
        })

    def _request(self, method: str, endpoint: str, params: Dict = None, data: Dict = None) -> Dict:
        url = f"{self.base_url}{endpoint}"

        response = self.session.request(
            method=method,
            url=url,
            params=params,
            json=data,
            timeout=30
        )

        result = response.json()

        if response.status_code >= 400:
            raise PixlpayError(
                message=result.get('message', 'Unknown error'),
                code=result.get('error', 'UNKNOWN'),
                status=response.status_code,
                errors=result.get('errors')
            )

        return result

    # Products
    def get_products(self, **params) -> Dict:
        return self._request('GET', '/products', params=params)

    def get_product(self, product_id: int) -> Dict:
        return self._request('GET', f'/products/{product_id}')

    # Orders
    def get_orders(self, **params) -> Dict:
        return self._request('GET', '/orders', params=params)

    def get_order(self, order_id: int) -> Dict:
        return self._request('GET', f'/orders/{order_id}')

    def fulfill_order(self, order_id: int) -> Dict:
        return self._request('POST', f'/orders/{order_id}/fulfill')

    # Customers
    def get_customers(self, **params) -> Dict:
        return self._request('GET', '/customers', params=params)

    def get_customer(self, customer_id: int) -> Dict:
        return self._request('GET', f'/customers/{customer_id}')

    # Analytics
    def get_revenue_analytics(self, **params) -> Dict:
        return self._request('GET', '/analytics/revenue', params=params)

    def get_sales_analytics(self, **params) -> Dict:
        return self._request('GET', '/analytics/sales', params=params)

    # Webhooks
    def get_webhooks(self) -> Dict:
        return self._request('GET', '/webhooks')

    def create_webhook(self, url: str, events: List[str]) -> Dict:
        return self._request('POST', '/webhooks', data={
            'url': url,
            'events': events
        })

    def delete_webhook(self, webhook_id: int) -> Dict:
        return self._request('DELETE', f'/webhooks/{webhook_id}')

Usage Examples

List Products

client = PixlpayClient(
    os.environ['PIXLPAY_STORE_URL'],
    os.environ['PIXLPAY_API_TOKEN']
)

try:
    response = client.get_products(is_active=True, per_page=50)

    for product in response['data']:
        print(f"{product['name']} - ${product['price']}")
except PixlpayError as e:
    print(f"Error: {e.message}")

Fulfill an Order

try:
    result = client.fulfill_order(123)
    print(f"Order {result['data']['order_number']} fulfilled!")
except PixlpayError as e:
    if e.code == 'ORDER_ALREADY_COMPLETED':
        print("Order was already completed")
    else:
        print(f"Error: {e.message}")

Get All Orders with Pagination

def get_all_orders(client: PixlpayClient, **filters) -> List[Dict]:
    all_orders = []
    page = 1

    while True:
        response = client.get_orders(page=page, per_page=100, **filters)
        all_orders.extend(response['data'])

        if page >= response['meta']['last_page']:
            break
        page += 1

    return all_orders

# Usage
paid_orders = get_all_orders(client, payment_status='paid')
print(f"Total paid orders: {len(paid_orders)}")

Flask Webhook Handler

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

app = Flask(__name__)

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

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

    # Verify signature
    if not verify_signature(payload, signature, secret):
        return jsonify({'error': 'Invalid signature'}), 401

    event = request.get_json()
    print(f"Received: {event['event_type']} ({event['id']})")

    # Process event
    try:
        process_event(event)
        return jsonify({'status': 'success'}), 200
    except Exception as e:
        print(f"Error: {e}")
        return jsonify({'error': str(e)}), 500

def process_event(event: Dict):
    event_type = event['event_type']
    data = event['data']

    handlers = {
        'order.paid': handle_order_paid,
        'order.refunded': handle_order_refunded,
        'subscription.created': handle_subscription_created,
        'subscription.cancelled': handle_subscription_cancelled,
    }

    handler = handlers.get(event_type)
    if handler:
        handler(data)
    else:
        print(f"Unhandled event: {event_type}")

def handle_order_paid(data: Dict):
    order_id = data['order_id']
    customer_email = data['customer_email']
    items = data['items']

    for item in items:
        deliver_item(customer_email, item)

    print(f"Delivered order {order_id} to {customer_email}")

def handle_order_refunded(data: Dict):
    print(f"Refund processed for order {data['order_id']}")

def handle_subscription_created(data: Dict):
    print(f"Subscription {data['plan_name']} created")

def handle_subscription_cancelled(data: Dict):
    print(f"Subscription {data['subscription_id']} cancelled")

def deliver_item(email: str, item: Dict):
    # Implement your delivery logic
    pass

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

Using with Celery

from celery import Celery

celery = Celery('pixlpay', broker='redis://localhost:6379/0')

@celery.task(bind=True, max_retries=3)
def process_webhook(self, event: Dict):
    try:
        process_event(event)
    except Exception as e:
        raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries))

# In webhook handler
@app.route('/webhooks/pixlpay', methods=['POST'])
def handle_webhook():
    # Verify signature...

    event = request.get_json()

    # Queue for async processing
    process_webhook.delay(event)

    return jsonify({'status': 'queued'}), 200

Type Hints

from typing import TypedDict, List, Optional
from datetime import datetime

class Product(TypedDict):
    id: int
    name: str
    description: str
    price: str
    type: str  # 'digital' | 'physical' | 'subscription'
    is_active: bool
    created_at: str
    updated_at: str

class OrderItem(TypedDict):
    id: int
    product_id: int
    product_name: str
    quantity: int
    price: str
    total: str

class Order(TypedDict):
    id: int
    order_number: str
    status: str
    payment_status: str
    total: str
    currency: str
    customer_email: str
    customer_name: str
    items: List[OrderItem]
    created_at: str
    updated_at: str

class WebhookEvent(TypedDict):
    id: str
    event_type: str
    created_at: str
    data: dict