Python SDK

The Torque Python SDK provides a simple and powerful way to integrate Torque's multi-product checkout into your Python applications. Built with modern Python practices, our SDK handles API interactions, webhook management, and data validation.

Installation

Pip Package

pip install torque-checkout

Poetry

poetry add torque-checkout

Requirements.txt

torque-checkout>=1.0.0

Configuration

Initialize SDK

from torque_checkout import TorqueCheckout

        torque = TorqueCheckout(
            business_id='your_business_id',
            api_key='your_api_key'
            # No baseUrl needed - automatically uses production endpoints
        )

Environment Variables

# .env file
        TORQUE_BUSINESS_ID=your_business_id
        TORQUE_API_KEY=your_api_key

import os
from torque_checkout import TorqueCheckout

torque = TorqueCheckout(
    business_id=os.getenv('TORQUE_BUSINESS_ID'),
    api_key=os.getenv('TORQUE_API_KEY')
)

API Methods

Checkout Management

Generate Checkout Link

# Basic checkout link generation
result = torque.generate_checkout_link(cart, customer_data)
checkout_url = result['checkoutUrl']

# With options
result = torque.generate_checkout_link(cart, customer_data, {
    'expiresIn': 3600,  # 1 hour
    'redirectUrl': 'https://yourdomain.com/thank-you',
    'webhookUrl': 'https://yourdomain.com/webhooks/torque',
    'metadata': {
        'orderId': 'order_123',
        'source': 'website'
    }
})

Retrieve Order

# Get order by ID
order = torque.get_order('order_123')

# Get order with specific fields
order = torque.get_order('order_123', {
    'fields': ['id', 'status', 'amount', 'customer']
})

List Orders

# Get all orders
orders = torque.list_orders()

# Get orders with filters
orders = torque.list_orders({
    'status': 'paid',
    'startDate': '2024-01-01',
    'endDate': '2024-12-31',
    'limit': 50,
    'offset': 0
})

Cancel Order

# Cancel an order
result = torque.cancel_order('order_123', {
    'reason': 'Customer requested cancellation'
})

Webhook Management

Create Webhook

# Create a new webhook
webhook = torque.create_webhook({
    'url': 'https://yourdomain.com/webhooks/torque',
    'events': ['order.created', 'order.paid', 'order.completed'],
    'secret': 'your_webhook_secret'
})

List Webhooks

# Get all webhooks
webhooks = torque.list_webhooks()

# Get webhooks by event type
webhooks = torque.list_webhooks({
    'event': 'order.paid'
})

Update Webhook

# Update webhook configuration
webhook = torque.update_webhook('webhook_123', {
    'url': 'https://newdomain.com/webhooks/torque',
    'events': ['order.created', 'order.paid', 'order.completed', 'order.cancelled']
})

Delete Webhook

# Delete a webhook
torque.delete_webhook('webhook_123')

Analytics

Business Overview

# Get business overview
overview = torque.get_business_overview({
    'period': 'month',  # 'day', 'week', 'month', 'year'
    'startDate': '2024-01-01',
    'endDate': '2024-12-31'
})

Revenue Analytics

# Get revenue analytics
revenue = torque.get_revenue_analytics({
    'period': 'month',
    'startDate': '2024-01-01',
    'endDate': '2024-12-31',
    'groupBy': 'day'  # 'day', 'week', 'month'
})

Order Analytics

# Get order analytics
orders = torque.get_order_analytics({
    'period': 'month',
    'startDate': '2024-01-01',
    'endDate': '2024-12-31',
    'metrics': ['count', 'value', 'conversion_rate']
})

Framework Integration

Django Integration

# settings.py
TORQUE_BUSINESS_ID = os.getenv('TORQUE_BUSINESS_ID')
TORQUE_API_KEY = os.getenv('TORQUE_API_KEY')
TORQUE_ENVIRONMENT = os.getenv('TORQUE_ENVIRONMENT', 'production')

# services/torque_service.py
from django.conf import settings
from torque_checkout import TorqueCheckout

class TorqueService:
    def __init__(self):
        self.torque = TorqueCheckout(
            business_id=settings.TORQUE_BUSINESS_ID,
            api_key=settings.TORQUE_API_KEY,
            environment=settings.TORQUE_ENVIRONMENT
        )
    
    def generate_checkout(self, cart, customer_data):
        return self.torque.generate_checkout_link(cart, customer_data)
    
    def get_order(self, order_id):
        return self.torque.get_order(order_id)

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import json
from .services.torque_service import TorqueService

torque_service = TorqueService()

@csrf_exempt
@require_http_methods(["POST"])
def generate_checkout(request):
    try:
        data = json.loads(request.body)
        cart = data.get('cart')
        customer_data = data.get('customerData')
        
        if not cart or not customer_data:
            return JsonResponse({'error': 'Missing cart or customer data'}, status=400)
        
        result = torque_service.generate_checkout(cart, customer_data)
        return JsonResponse(result)
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

Flask Integration

# app.py
from flask import Flask, request, jsonify
from torque_checkout import TorqueCheckout
import os

app = Flask(__name__)

torque = TorqueCheckout(
    business_id=os.getenv('TORQUE_BUSINESS_ID'),
    api_key=os.getenv('TORQUE_API_KEY')
    # No environment configuration needed
)

@app.route('/api/checkout', methods=['POST'])
def generate_checkout():
    try:
        data = request.get_json()
        cart = data.get('cart')
        customer_data = data.get('customerData')
        
        if not cart or not customer_data:
            return jsonify({'error': 'Missing cart or customer data'}), 400
        
        result = torque.generate_checkout_link(cart, customer_data)
        return jsonify(result)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/webhooks/torque', methods=['POST'])
def webhook():
    try:
        # Verify webhook signature
        signature = request.headers.get('X-Torque-Signature', '')
        payload = request.get_data(as_text=True)
        
        if not torque.verify_webhook_signature(payload, signature):
            return jsonify({'error': 'Invalid signature'}), 401
        
        data = request.get_json()
        event = data.get('event')
        event_data = data.get('data', {})
        
        # Process webhook event
        process_webhook_event(event, event_data)
        
        return jsonify({'success': True})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

def process_webhook_event(event, event_data):
    if event == 'order.paid':
        handle_order_paid(event_data)
    elif event == 'order.completed':
        handle_order_completed(event_data)
    elif event == 'order.cancelled':
        handle_order_cancelled(event_data)

def handle_order_paid(data):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been paid")
    
    # Update your system with the order status
    # Send confirmation emails, update inventory, etc.

if __name__ == '__main__':
    app.run(debug=True)

FastAPI Integration

# main.py
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
from typing import Optional, List
from torque_checkout import TorqueCheckout
import os

app = FastAPI()

torque = TorqueCheckout(
    business_id=os.getenv('TORQUE_BUSINESS_ID'),
    api_key=os.getenv('TORQUE_API_KEY')
    # No environment configuration needed
)

class CartItem(BaseModel):
    productId: str
    quantity: int
    price: float
    variant: Optional[str] = None
    metadata: Optional[dict] = None

class Cart(BaseModel):
    items: List[CartItem]
    currency: str = "USD"
    discounts: Optional[List[dict]] = None

class CustomerData(BaseModel):
    email: str
    firstName: str
    lastName: str
    phone: Optional[str] = None
    address: Optional[dict] = None

class CheckoutRequest(BaseModel):
    cart: Cart
    customerData: CustomerData

@app.post("/api/checkout")
async def generate_checkout(request: CheckoutRequest):
    try:
        result = torque.generate_checkout_link(
            request.cart.dict(),
            request.customerData.dict()
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/webhooks/torque")
async def webhook(
    request: dict,
    x_torque_signature: str = Header(None)
):
    try:
        if not x_torque_signature:
            raise HTTPException(status_code=401, detail="Missing signature")
        
        # Verify webhook signature
        if not torque.verify_webhook_signature(
            str(request),
            x_torque_signature
        ):
            raise HTTPException(status_code=401, detail="Invalid signature")
        
        event = request.get('event')
        event_data = request.get('data', {})
        
        # Process webhook event
        await process_webhook_event(event, event_data)
        
        return {"success": True}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

async def process_webhook_event(event: str, event_data: dict):
    if event == 'order.paid':
        await handle_order_paid(event_data)
    elif event == 'order.completed':
        await handle_order_completed(event_data)
    elif event == 'order.cancelled':
        await handle_order_cancelled(event_data)

async def handle_order_paid(data: dict):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been paid")
    
    # Update your system with the order status
    # Send confirmation emails, update inventory, etc.

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Webhook Handling

Standalone Webhook Handler

# webhook_handler.py
from flask import Flask, request, jsonify
from torque_checkout import TorqueCheckout
import os
import json

app = Flask(__name__)

torque = TorqueCheckout(
    business_id=os.getenv('TORQUE_BUSINESS_ID'),
    api_key=os.getenv('TORQUE_API_KEY')
)

webhook_secret = os.getenv('TORQUE_WEBHOOK_SECRET')

@app.route('/webhooks/torque', methods=['POST'])
def webhook():
    try:
        # Get webhook data
        payload = request.get_data(as_text=True)
        signature = request.headers.get('X-Torque-Signature', '')
        
        # Verify webhook signature
        if not torque.verify_webhook_signature(payload, signature, webhook_secret):
            return jsonify({'error': 'Invalid signature'}), 401
        
        data = request.get_json()
        event = data.get('event')
        event_data = data.get('data', {})
        
        # Process webhook event
        process_webhook_event(event, event_data)
        
        return jsonify({'success': True})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

def process_webhook_event(event, event_data):
    if event == 'order.paid':
        handle_order_paid(event_data)
    elif event == 'order.completed':
        handle_order_completed(event_data)
    elif event == 'order.cancelled':
        handle_order_cancelled(event_data)

def handle_order_paid(data):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been paid")
    
    # Update your system with the order status
    # Send confirmation emails, update inventory, etc.

def handle_order_completed(data):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been completed")
    
    # Handle order completion

def handle_order_cancelled(data):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been cancelled")
    
    # Handle order cancellation

if __name__ == '__main__':
    app.run(debug=True, port=5000)

Async Webhook Handler

# async_webhook_handler.py
import asyncio
import aiohttp
from aiohttp import web
from torque_checkout import TorqueCheckout
import os
import json

torque = TorqueCheckout(
    business_id=os.getenv('TORQUE_BUSINESS_ID'),
    api_key=os.getenv('TORQUE_API_KEY')
)

webhook_secret = os.getenv('TORQUE_WEBHOOK_SECRET')

async def webhook_handler(request):
    try:
        # Get webhook data
        payload = await request.text()
        signature = request.headers.get('X-Torque-Signature', '')
        
        # Verify webhook signature
        if not torque.verify_webhook_signature(payload, signature, webhook_secret):
            return web.json_response({'error': 'Invalid signature'}, status=401)
        
        data = await request.json()
        event = data.get('event')
        event_data = data.get('data', {})
        
        # Process webhook event asynchronously
        asyncio.create_task(process_webhook_event(event, event_data))
        
        return web.json_response({'success': True})
    except Exception as e:
        return web.json_response({'error': str(e)}, status=500)

async def process_webhook_event(event, event_data):
    if event == 'order.paid':
        await handle_order_paid(event_data)
    elif event == 'order.completed':
        await handle_order_completed(event_data)
    elif event == 'order.cancelled':
        await handle_order_cancelled(event_data)

async def handle_order_paid(data):
    order_id = data.get('orderId')
    print(f"Order {order_id} has been paid")
    
    # Update your system with the order status
    # Send confirmation emails, update inventory, etc.
    
    # Example: Send async notification
    await send_notification(f"Order {order_id} has been paid")

async def send_notification(message):
    # Simulate async notification
    await asyncio.sleep(1)
    print(f"Notification sent: {message}")

app = web.Application()
app.router.add_post('/webhooks/torque', webhook_handler)

if __name__ == '__main__':
    web.run_app(app, port=5000)

Testing

Test Environment

# Use sandbox environment for testing
torque = TorqueCheckout(
    business_id='test_business_123',
    api_key='test_api_key_456',
    environment='sandbox'
)

# Test checkout generation
result = torque.generate_checkout_link({
    'items': [
        {
            'productId': 'test_prod_1',
            'quantity': 1,
            'price': 9.99
        }
    ]
}, {
    'email': 'test@example.com',
    'firstName': 'Test',
    'lastName': 'User'
})

Mock Data

# Sample cart data for testing
test_cart = {
    'items': [
        {
            'productId': 'prod_1',
            'quantity': 2,
            'price': 29.99,
            'variant': 'Blue - Large',
            'metadata': {
                'sku': 'BLUE-L-001',
                'name': 'Premium T-Shirt'
            }
        },
        {
            'productId': 'prod_2',
            'quantity': 1,
            'price': 19.99,
            'variant': 'Red - Medium',
            'metadata': {
                'sku': 'RED-M-002',
                'name': 'Casual Shirt'
            }
        }
    ],
    'currency': 'USD',
    'discounts': [
        {
            'code': 'SAVE10',
            'amount': 5.00,
            'type': 'fixed'
        }
    ]
}

# Sample customer data
test_customer = {
    'email': 'customer@example.com',
    'firstName': 'John',
    'lastName': 'Doe',
    'phone': '+1234567890',
    'address': {
        'street': '123 Main St',
        'city': 'New York',
        'state': 'NY',
        'postalCode': '10001',
        'country': 'US'
    }
}

Error Handling

Try-Except with Retry Logic

import asyncio
from typing import Optional, Dict, Any

async def generate_checkout_with_retry(
    torque, 
    cart: Dict[str, Any], 
    customer_data: Dict[str, Any], 
    max_retries: int = 3
) -> Optional[Dict[str, Any]]:
    for attempt in range(1, max_retries + 1):
        try:
            result = torque.generate_checkout_link(cart, customer_data)
            return result
        except Exception as e:
            if attempt == max_retries:
                raise e
            
            # Wait before retrying (exponential backoff)
            delay = 2 ** attempt
            print(f"Attempt {attempt} failed, retrying in {delay}s...")
            await asyncio.sleep(delay)
    
    return None

Error Types

from torque_checkout import TorqueError

try:
    result = torque.generate_checkout_link(cart, customer_data)
except TorqueError as e:
    if e.code == 'INVALID_CART':
        print(f"Cart data is invalid: {e.message}")
    elif e.code == 'INSUFFICIENT_FUNDS':
        print("Insufficient funds for transaction")
    elif e.code == 'RATE_LIMIT_EXCEEDED':
        print("Rate limit exceeded, please wait")
    else:
        print(f"Torque error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Next Steps

• JavaScript SDK: JavaScript/Node.js integration guide
• PHP SDK: PHP integration guide
• API Reference: Raw API documentation
• Webhook Guide: Webhook setup and handling
• Error Handling: Error codes and troubleshooting

---

Need help with Python integration? Contact our support team at hello@torque.fi or check our SDK troubleshooting guide.