Python Examples

Complete code examples for using the TakeTheme API with Python.

Setup

Using Requests

pip install requests

import os
import requests
from typing import Optional, Dict, Any, Iterator

API_KEY = os.environ.get('TAKETHEME_API_KEY')
BASE_URL = 'https://api.taketheme.com/api/v1'

class TakeThemeClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            'tt-api-key': f'{api_key}',
            'Content-Type': 'application/json'
        })

    def request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        response = self.session.request(
            method,
            f'{BASE_URL}{endpoint}',
            **kwargs
        )

        if not response.ok:
            error_data = response.json()
            raise TakeThemeError(error_data, response.status_code)

        return response.json()

    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]:
        return self.request('GET', endpoint, params=params)

    def post(self, endpoint: str, data: Dict) -> Dict[str, Any]:
        return self.request('POST', endpoint, json=data)

    def patch(self, endpoint: str, data: Dict) -> Dict[str, Any]:
        return self.request('PATCH', endpoint, json=data)

    def delete(self, endpoint: str) -> None:
        self.request('DELETE', endpoint)


class TakeThemeError(Exception):
    def __init__(self, error: Dict, status_code: int):
        self.message = error.get('message') or error.get('name', 'Unknown error')
        self.status = error.get('status', status_code)
        self.code = error.get('code')  # Only present on some errors like rate limits
        super().__init__(self.message)


# Initialize client
client = TakeThemeClient(API_KEY)

Products

List All Products

# Using requests
response = client.get('/products', params={'limit': 25})
products = response['data']
pagination = response['pagination']

print(f"Found {pagination['total_count']} products")
for product in products:
    print(f"- {product['title']} (${product['price']})")

Get a Single Product

product = client.get('/products/prod_abc123')['data']
print(f"{product['title']}: {product['description']}")

Create a Product

new_product = client.post('/products', {
    'title': 'Premium Hoodie',
    'description': 'Warm and comfortable hoodie',
    'price': 59.99,
    'currency': 'USD',
    'status': 'active',
    'variants': [
        {'title': 'Small', 'sku': 'HOODIE-S', 'price': 59.99, 'inventory_quantity': 50},
        {'title': 'Medium', 'sku': 'HOODIE-M', 'price': 59.99, 'inventory_quantity': 75},
        {'title': 'Large', 'sku': 'HOODIE-L', 'price': 59.99, 'inventory_quantity': 60},
    ],
    'images': [
        {'src': 'https://example.com/hoodie.jpg', 'alt': 'Premium Hoodie'}
    ]
})['data']

print(f"Created product: {new_product['id']}")

Update a Product

updated = client.patch('/products/prod_abc123', {
    'price': 64.99,
    'tags': ['featured', 'winter-collection']
})['data']

Delete a Product

client.delete('/products/prod_abc123')

Orders

List Orders

# Recent pending orders
orders = client.get('/orders', params={
    'status': 'pending',
    'limit': 50
})['data']

# Orders with date filtering
from datetime import datetime, timedelta

week_ago = (datetime.now() - timedelta(days=7)).isoformat()
recent_orders = client.get('/orders', params={
    'created_at_min': week_ago
})['data']

Create an Order

order = client.post('/orders', {
    'customer': {
        'email': 'customer@example.com',
        'first_name': 'John',
        'last_name': 'Doe'
    },
    'line_items': [
        {
            'variant_id': 'var_abc123',
            'quantity': 2,
            'price': 29.99
        }
    ],
    'shipping_address': {
        'first_name': 'John',
        'last_name': 'Doe',
        'address1': '123 Main Street',
        'city': 'New York',
        'province': 'NY',
        'postal_code': '10001',
        'country': 'US',
        'phone': '+1-555-123-4567'
    },
    'shipping_line': {
        'title': 'Standard Shipping',
        'price': 5.99
    }
})['data']

print(f"Order created: {order['order_number']}")

Fulfill an Order

fulfillment = client.post(f'/orders/{order_id}/fulfillments', {
    'line_items': [
        {'id': 'li_abc123', 'quantity': 2}
    ],
    'tracking_number': '1Z999AA10123456784',
    'tracking_company': 'UPS',
    'notify_customer': True
})['data']

Cancel an Order

cancelled = client.post(f'/orders/{order_id}/cancel', {
    'reason': 'customer_request',
    'restock': True,
    'notify_customer': True
})['data']

Customers

Search Customers

# Find by email
customers = client.get('/customers', params={
    'email': 'john@example.com'
})['data']

# Find VIP customers
vip_customers = client.get('/customers', params={
    'tags': 'vip',
    'orders_count_min': 10
})['data']

Create a Customer

customer = client.post('/customers', {
    'email': 'newcustomer@example.com',
    'first_name': 'Jane',
    'last_name': 'Smith',
    'phone': '+1-555-987-6543',
    'addresses': [
        {
            'address1': '456 Oak Avenue',
            'city': 'Los Angeles',
            'province': 'CA',
            'postal_code': '90001',
            'country': 'US',
            'default': True
        }
    ],
    'tags': ['newsletter', 'loyalty-program'],
    'accepts_marketing': True
})['data']

Inventory

Update Stock Levels

# Set absolute quantity
client.post('/inventory/var_abc123', {
    'location_id': 'loc_main',
    'available': 100
})

# Adjust by delta
client.post('/inventory/var_abc123/adjust', {
    'location_id': 'loc_main',
    'adjustment': -5,
    'reason': 'Damaged stock'
})

Webhooks

Register a Webhook

webhook = client.post('/webhooks', {
    'topic': 'orders/created',
    'address': 'https://your-app.com/webhooks/orders',
    'format': 'json'
})['data']

print(f"Webhook registered: {webhook['id']}")

Verify Webhook Signatures (Flask)

import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
WEBHOOK_SECRET = os.environ['TAKETHEME_WEBHOOK_SECRET']

def verify_webhook(payload: bytes, signature: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).digest()

    return hmac.compare_digest(
        expected,
        base64.b64decode(signature)
    )

@app.route('/webhooks/orders', methods=['POST'])
def handle_order_webhook():
    signature = request.headers.get('X-TakeTheme-Signature')

    if not signature or not verify_webhook(request.data, signature):
        abort(401)

    event = request.json
    print(f"Order created: {event['data']['order_number']}")

    return 'OK', 200

Pagination Helper

from typing import Iterator, Dict, Any

def paginate(endpoint: str, params: Optional[Dict] = None) -> Iterator[Dict[str, Any]]:
    """Generator that yields items from all pages."""
    params = params or {}
    params['limit'] = 100
    cursor = None

    while True:
        if cursor:
            params['cursor'] = cursor
            params['direction'] = 'next'

        response = client.get(endpoint, params=params)
        data = response['data']
        pagination = response['pagination']

        for item in data:
            yield item

        if not pagination.get('has_more'):
            break

        cursor = pagination.get('next_cursor')


# Usage
for product in paginate('/products'):
    print(product['title'])

# Collect all products
all_products = list(paginate('/products'))
print(f"Total: {len(all_products)} products")

Error Handling

from typing import Dict

class TakeThemeError(Exception):
    def __init__(self, error: Dict, status_code: int):
        self.message = error.get('message') or error.get('name', 'Unknown error')
        self.status = error.get('status', status_code)
        self.code = error.get('code')  # Only present on some errors like rate limits
        super().__init__(self.message)


class RateLimitError(TakeThemeError):
    def __init__(self, error: Dict, status_code: int, retry_after: int):
        super().__init__(error, status_code)
        self.retry_after = retry_after


def handle_response(response: requests.Response) -> Dict:
    if response.ok:
        return response.json()

    error_data = response.json()

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        raise RateLimitError(error_data, response.status_code, retry_after)
    else:
        raise TakeThemeError(error_data, response.status_code)


# Usage
try:
    client.post('/products', {'title': ''})  # Invalid
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
    time.sleep(e.retry_after)
except TakeThemeError as e:
    if e.status == 400:
        print(f"Validation error: {e.message}")
    elif e.status == 401:
        print("Authentication failed")
    elif e.status == 403:
        print(f"Permission denied: {e.message}")
    else:
        print(f"Error: {e.message}")

Complete Example: Export Orders to CSV

import csv
import os
from datetime import datetime, timedelta
from typing import Iterator, Dict

API_KEY = os.environ['TAKETHEME_API_KEY']

def get_recent_orders(days: int = 30) -> Iterator[Dict]:
    """Fetch all orders from the last N days."""
    since = (datetime.now() - timedelta(days=days)).isoformat()

    for order in paginate('/orders', {'created_at_min': since}):
        yield order


def export_orders_to_csv(filename: str, days: int = 30):
    """Export recent orders to a CSV file."""
    fieldnames = [
        'order_number',
        'created_at',
        'customer_email',
        'total',
        'status',
        'fulfillment_status',
        'items_count'
    ]

    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

        for order in get_recent_orders(days):
            writer.writerow({
                'order_number': order['order_number'],
                'created_at': order['created_at'],
                'customer_email': order.get('customer', {}).get('email', ''),
                'total': order['total'],
                'status': order['status'],
                'fulfillment_status': order.get('fulfillment_status', 'unfulfilled'),
                'items_count': len(order.get('line_items', []))
            })

    print(f"Exported orders to {filename}")


if __name__ == '__main__':
    export_orders_to_csv('orders_export.csv', days=30)

Async Example (aiohttp)

import asyncio
import aiohttp
import os

API_KEY = os.environ['TAKETHEME_API_KEY']
BASE_URL = 'https://api.taketheme.com/api/v1'

async def fetch_products():
    async with aiohttp.ClientSession() as session:
        headers = {
            'tt-api-key': f'{API_KEY}',
            'Content-Type': 'application/json'
        }

        async with session.get(f'{BASE_URL}/products', headers=headers) as response:
            data = await response.json()
            return data['data']


async def fetch_multiple_products(product_ids: list):
    async with aiohttp.ClientSession() as session:
        headers = {
            'tt-api-key': f'{API_KEY}',
            'Content-Type': 'application/json'
        }

        tasks = [
            session.get(f'{BASE_URL}/products/{pid}', headers=headers)
            for pid in product_ids
        ]

        responses = await asyncio.gather(*tasks)
        products = []

        for response in responses:
            data = await response.json()
            products.append(data['data'])

        return products


# Usage
async def main():
    products = await fetch_products()
    print(f"Found {len(products)} products")

    # Fetch multiple products concurrently
    ids = [p['id'] for p in products[:5]]
    details = await fetch_multiple_products(ids)
    for p in details:
        print(f"- {p['title']}")


asyncio.run(main())