Python SDK

The official Python SDK for Unified Commerce Platform. Built with modern Python features, featuring full async support, comprehensive type hints, and Pydantic models for data validation.

Version: 1.8.3 | License: MIT | Python: 3.8+ | Size: ~180KB

Installation

Install the SDK using pip:

# Using pip
pip install unified-commerce

# Using poetry
poetry add unified-commerce

# Using pipenv
pipenv install unified-commerce

# With async extras
pip install unified-commerce[async]

Requirements

Python 3.8 or higher
pip 21.0 or higher

Optional Dependencies

# Async support (httpx, aiohttp)
pip install unified-commerce[async]

# Django integration
pip install unified-commerce[django]

# FastAPI integration
pip install unified-commerce[fastapi]

# All extras
pip install unified-commerce[all]

Quick Start

Basic Setup

import os
from unified_commerce import UnifiedCommerce

# Initialize the client
client = UnifiedCommerce(
    api_key=os.environ['UNIFIED_COMMERCE_API_KEY'],
    environment='production'  # or 'sandbox' for testing
)

# Query products
products = client.products.list(
    category='electronics',
    limit=10
)

for product in products:
    print(f"{product.name}: ${product.price}")

Async Setup

import asyncio
import os
from unified_commerce import AsyncUnifiedCommerce

async def main():
    # Initialize async client
    client = AsyncUnifiedCommerce(
        api_key=os.environ['UNIFIED_COMMERCE_API_KEY'],
        environment='production'
    )

    # Query products asynchronously
    products = await client.products.list(
        category='electronics',
        limit=10
    )

    for product in products:
        print(f"{product.name}: ${product.price}")

    # Close the client
    await client.close()

if __name__ == '__main__':
    asyncio.run(main())

Context Manager

from unified_commerce import UnifiedCommerce

# Automatically handles client lifecycle
with UnifiedCommerce(api_key=os.environ['UNIFIED_COMMERCE_API_KEY']) as client:
    products = client.products.list(limit=10)
    print(products)

# Async context manager
async with AsyncUnifiedCommerce(api_key=os.environ['UNIFIED_COMMERCE_API_KEY']) as client:
    products = await client.products.list(limit=10)
    print(products)

Configuration

Client Options

from unified_commerce import UnifiedCommerce

client = UnifiedCommerce(
    # Required: Your API key
    api_key='uc_live_xxxxxxxxxxxxx',

    # Environment: 'production' or 'sandbox'
    environment='production',

    # Custom API endpoint
    api_url='https://api.custom-domain.com',

    # Request timeout in seconds (default: 30)
    timeout=60,

    # Maximum retry attempts (default: 3)
    max_retries=5,

    # Retry backoff factor (default: 2)
    retry_backoff=2,

    # Enable debug logging (default: False)
    debug=True,

    # Custom headers for all requests
    headers={
        'X-Custom-Header': 'value'
    },

    # HTTP session (for connection pooling)
    session=custom_session,

    # Proxy configuration
    proxies={
        'http': 'http://proxy.example.com:8080',
        'https': 'https://proxy.example.com:8080'
    }
)

Environment Variables

Create a .env file in your project:

# .env
UNIFIED_COMMERCE_API_KEY=uc_live_xxxxxxxxxxxxx
UNIFIED_COMMERCE_ENVIRONMENT=production
UNIFIED_COMMERCE_TIMEOUT=60

Load environment variables:

from dotenv import load_dotenv
from unified_commerce import UnifiedCommerce

load_dotenv()

client = UnifiedCommerce.from_env()  # Loads from environment

Authentication

API Key Authentication

from unified_commerce import UnifiedCommerce

# Method 1: Direct initialization
client = UnifiedCommerce(api_key='uc_live_xxxxxxxxxxxxx')

# Method 2: From environment variable
client = UnifiedCommerce(api_key=os.environ['UNIFIED_COMMERCE_API_KEY'])

# Method 3: Using configuration object
from unified_commerce import Config

config = Config(
    api_key='uc_live_xxxxxxxxxxxxx',
    environment='production'
)
client = UnifiedCommerce(config=config)

Per-Request Authentication

# Override API key for specific requests
products = client.products.list(
    limit=10,
    api_key='different_api_key'
)

Service Account (Coming Soon)

from unified_commerce import UnifiedCommerce

client = UnifiedCommerce(
    service_account={
        'type': 'service_account',
        'project_id': 'your-project-id',
        'private_key': 'your-private-key',
        'client_email': 'service@project.iam.gserviceaccount.com'
    }
)

Core Concepts

Products

from unified_commerce import UnifiedCommerce
from unified_commerce.models import Product, ProductCreate, ProductUpdate

client = UnifiedCommerce(api_key=os.environ['UNIFIED_COMMERCE_API_KEY'])

# List products with filters
products = client.products.list(
    category='electronics',
    min_price=100,
    max_price=1000,
    in_stock=True,
    limit=20,
    offset=0,
    sort_by='price',
    sort_order='asc'
)

# Get a single product
product = client.products.get('product-123')

# Create a product
new_product = client.products.create(
    ProductCreate(
        name='Wireless Headphones',
        description='Premium noise-cancelling headphones',
        price=299.99,
        category='electronics',
        sku='WH-1000XM5',
        inventory={
            'quantity': 100,
            'track_inventory': True
        }
    )
)

# Update a product
updated = client.products.update(
    'product-123',
    ProductUpdate(
        price=279.99,
        inventory={'quantity': 150}
    )
)

# Delete a product
client.products.delete('product-123')

# Search products
results = client.products.search(
    query='wireless headphones',
    filters={'category': 'electronics'}
)

Orders

from unified_commerce.models import OrderCreate, OrderItem, Customer, Address

# Create an order
order = client.orders.create(
    OrderCreate(
        items=[
            OrderItem(
                product_id='product-123',
                quantity=2,
                price=299.99
            ),
            OrderItem(
                product_id='product-456',
                quantity=1,
                price=149.99
            )
        ],
        customer=Customer(
            id='customer-789',
            email='customer@example.com',
            name='John Doe'
        ),
        shipping_address=Address(
            street='123 Main St',
            city='San Francisco',
            state='CA',
            postal_code='94105',
            country='US'
        ),
        payment_method={
            'type': 'card',
            'token': 'pm_xxxxxxxxxxxxx'
        }
    )
)

# List orders
orders = client.orders.list(
    customer_id='customer-789',
    status='pending',
    limit=20
)

# Get order details
order_details = client.orders.get('order-123')

# Update order status
client.orders.update(
    'order-123',
    status='shipped',
    tracking={
        'carrier': 'UPS',
        'tracking_number': '1Z999AA10123456784'
    }
)

# Cancel an order
client.orders.cancel(
    'order-123',
    reason='Customer requested cancellation'
)

Customers

from unified_commerce.models import CustomerCreate, CustomerUpdate

# Create a customer
customer = client.customers.create(
    CustomerCreate(
        email='customer@example.com',
        name='John Doe',
        phone='+1234567890',
        addresses=[{
            'type': 'shipping',
            'street': '123 Main St',
            'city': 'San Francisco',
            'state': 'CA',
            'postal_code': '94105',
            'country': 'US'
        }]
    )
)

# List customers
customers = client.customers.list(
    search='john',
    limit=20
)

# Update customer
client.customers.update(
    'customer-123',
    CustomerUpdate(phone='+9876543210')
)

# Get customer with orders
customer_with_orders = client.customers.get(
    'customer-123',
    include=['orders', 'addresses']
)

Inventory

from unified_commerce.models import InventoryUpdate

# Get inventory for a product
inventory = client.inventory.get('product-123')

# Update inventory
client.inventory.update(
    'product-123',
    InventoryUpdate(
        quantity=100,
        location='warehouse-1',
        reserved_quantity=5
    )
)

# Bulk update
client.inventory.bulk_update([
    {'product_id': 'product-123', 'quantity': 100},
    {'product_id': 'product-456', 'quantity': 50}
])

# Track inventory movements
movements = client.inventory.movements(
    'product-123',
    start_date='2024-01-01',
    end_date='2024-01-31'
)

# Reserve inventory
reservation = client.inventory.reserve(
    'product-123',
    quantity=5,
    expires_at='2024-12-31T23:59:59Z'
)

Async Support

Async Client

import asyncio
from unified_commerce import AsyncUnifiedCommerce

async def fetch_products():
    async with AsyncUnifiedCommerce(
        api_key=os.environ['UNIFIED_COMMERCE_API_KEY']
    ) as client:
        # Concurrent requests
        products, orders, customers = await asyncio.gather(
            client.products.list(limit=10),
            client.orders.list(limit=10),
            client.customers.list(limit=10)
        )

        return products, orders, customers

# Run async function
products, orders, customers = asyncio.run(fetch_products())

Async Iteration

async def process_all_products():
    async with AsyncUnifiedCommerce(
        api_key=os.environ['UNIFIED_COMMERCE_API_KEY']
    ) as client:
        # Async iterator for pagination
        async for product in client.products.iterate(category='electronics'):
            print(f"Processing {product.name}")
            await process_product(product)

Async Context Managers

from unified_commerce import AsyncUnifiedCommerce

class ProductProcessor:
    def __init__(self, api_key: str):
        self.client = AsyncUnifiedCommerce(api_key=api_key)

    async def __aenter__(self):
        await self.client.__aenter__()
        return self

    async def __aexit__(self, *args):
        await self.client.__aexit__(*args)

    async def process(self):
        products = await self.client.products.list(limit=100)
        # Process products
        return products

# Use it
async with ProductProcessor(api_key=os.environ['UNIFIED_COMMERCE_API_KEY']) as processor:
    results = await processor.process()

Django Integration

Settings Configuration

# settings.py
UNIFIED_COMMERCE = {
    'API_KEY': os.environ['UNIFIED_COMMERCE_API_KEY'],
    'ENVIRONMENT': 'production',
    'TIMEOUT': 30,
    'MAX_RETRIES': 3,
    'DEBUG': DEBUG
}

Model Integration

# models.py
from django.db import models
from unified_commerce.django import UnifiedCommerceField

class Product(models.Model):
    name = models.CharField(max_length=255)
    uc_product = UnifiedCommerceField()  # Syncs with Unified Commerce

    def save(self, *args, **kwargs):
        # Auto-sync to Unified Commerce
        if not self.uc_product:
            from unified_commerce.django import get_client
            client = get_client()
            uc_product = client.products.create({
                'name': self.name,
                # ... other fields
            })
            self.uc_product = uc_product.id
        super().save(*args, **kwargs)

Views Integration

# views.py
from django.http import JsonResponse
from unified_commerce.django import get_client

def product_list(request):
    client = get_client()
    products = client.products.list(
        category=request.GET.get('category'),
        limit=20
    )

    return JsonResponse({
        'products': [p.dict() for p in products]
    })

FastAPI Integration

Setup

# main.py
from fastapi import FastAPI, Depends
from unified_commerce.fastapi import UnifiedCommerceClient, get_client

app = FastAPI()

# Initialize at startup
@app.on_event("startup")
async def startup():
    await UnifiedCommerceClient.initialize(
        api_key=os.environ['UNIFIED_COMMERCE_API_KEY']
    )

@app.on_event("shutdown")
async def shutdown():
    await UnifiedCommerceClient.close()

Dependency Injection

from fastapi import APIRouter, Depends
from unified_commerce.fastapi import get_client
from unified_commerce import AsyncUnifiedCommerce

router = APIRouter()

@router.get("/products")
async def list_products(
    client: AsyncUnifiedCommerce = Depends(get_client)
):
    products = await client.products.list(limit=20)
    return {"products": products}

@router.post("/orders")
async def create_order(
    order_data: OrderCreate,
    client: AsyncUnifiedCommerce = Depends(get_client)
):
    order = await client.orders.create(order_data)
    return {"order": order}

Background Tasks

from fastapi import BackgroundTasks
from unified_commerce.fastapi import get_client

async def sync_inventory(product_id: str):
    client = await get_client()
    inventory = await client.inventory.get(product_id)
    # Update local database
    await update_local_inventory(inventory)

@router.post("/products/{product_id}/sync")
async def sync_product_inventory(
    product_id: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(sync_inventory, product_id)
    return {"status": "syncing"}

Error Handling

Exception Types

from unified_commerce.exceptions import (
    UnifiedCommerceError,
    AuthenticationError,
    ValidationError,
    NotFoundError,
    RateLimitError,
    NetworkError,
    ServerError
)

try:
    product = client.products.get('invalid-id')
except NotFoundError as e:
    print(f"Product not found: {e.message}")
except AuthenticationError as e:
    print(f"Invalid API key: {e.message}")
except RateLimitError as e:
    print(f"Rate limit exceeded. Retry after: {e.retry_after} seconds")
except ValidationError as e:
    print(f"Validation errors: {e.errors}")
    for field, errors in e.errors.items():
        print(f"  {field}: {', '.join(errors)}")
except UnifiedCommerceError as e:
    print(f"API error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Retry Decorator

from unified_commerce.decorators import retry

@retry(max_attempts=5, backoff=2)
def fetch_products():
    return client.products.list(limit=100)

# With custom retry logic
@retry(
    max_attempts=3,
    exceptions=(NetworkError, ServerError),
    on_retry=lambda e, attempt: print(f"Retry {attempt}: {e}")
)
async def fetch_orders():
    return await async_client.orders.list(limit=100)

Webhooks

Webhook Verification

from unified_commerce.webhooks import verify_webhook
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhooks/unified-commerce', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Unified-Commerce-Signature')
    payload = request.get_data()

    try:
        event = verify_webhook(
            payload,
            signature,
            os.environ['WEBHOOK_SECRET']
        )

        # Handle the verified event
        if event.type == 'order.created':
            handle_order_created(event.data)
        elif event.type == 'order.updated':
            handle_order_updated(event.data)
        elif event.type == 'inventory.low':
            handle_low_inventory(event.data)

        return jsonify({'received': True})

    except Exception as e:
        return jsonify({'error': str(e)}), 400

def handle_order_created(order):
    print(f"New order created: {order.id}")
    # Process order

Webhook Event Handlers

from unified_commerce.webhooks import WebhookHandler

handler = WebhookHandler(secret=os.environ['WEBHOOK_SECRET'])

@handler.on('order.created')
def handle_order_created(event):
    order = event.data
    print(f"Order created: {order.id}")
    send_confirmation_email(order)

@handler.on('payment.failed')
def handle_payment_failed(event):
    payment = event.data
    print(f"Payment failed: {payment.id}")
    notify_customer(payment)

@handler.on('inventory.low')
async def handle_low_inventory(event):
    inventory = event.data
    print(f"Low inventory alert: {inventory.product_id}")
    await reorder_stock(inventory)

# Process webhook
@app.route('/webhooks/unified-commerce', methods=['POST'])
def webhook():
    try:
        handler.handle(
            request.get_data(),
            request.headers.get('X-Unified-Commerce-Signature')
        )
        return jsonify({'received': True})
    except Exception as e:
        return jsonify({'error': str(e)}), 400

Advanced Features

GraphQL Queries

from unified_commerce.graphql import gql

# Custom GraphQL query
CUSTOM_QUERY = gql("""
    query GetProductsWithReviews($category: String!) {
        products(category: $category) {
            id
            name
            price
            reviews {
                id
                rating
                comment
                author {
                    name
                }
            }
        }
    }
""")

result = client.query(
    CUSTOM_QUERY,
    variables={'category': 'electronics'}
)

products = result['products']

Batch Operations

# Batch create products
products = client.products.batch_create([
    {'name': 'Product 1', 'price': 99.99},
    {'name': 'Product 2', 'price': 149.99},
    {'name': 'Product 3', 'price': 199.99}
])

# Batch update
client.products.batch_update([
    {'id': 'product-1', 'price': 89.99},
    {'id': 'product-2', 'price': 139.99}
])

# Batch delete
client.products.batch_delete(['product-1', 'product-2', 'product-3'])

Pagination

# Offset-based pagination
def fetch_all_products():
    all_products = []
    offset = 0
    limit = 100

    while True:
        products = client.products.list(
            limit=limit,
            offset=offset
        )

        all_products.extend(products.data)

        if len(products.data) < limit:
            break

        offset += limit

    return all_products

# Cursor-based pagination
def fetch_all_products_cursor():
    all_products = []
    cursor = None

    while True:
        response = client.products.list(
            limit=100,
            cursor=cursor
        )

        all_products.extend(response.data)

        if not response.next_cursor:
            break

        cursor = response.next_cursor

    return all_products

# Using iterator
for product in client.products.iterate(category='electronics'):
    print(product.name)

File Uploads

# Upload product image
with open('product-image.jpg', 'rb') as f:
    image = client.products.upload_image(
        'product-123',
        file=f,
        alt='Product image',
        position=0
    )

# Multiple images
images = [
    ('image1.jpg', 'Front view'),
    ('image2.jpg', 'Side view'),
    ('image3.jpg', 'Back view')
]

for filename, alt in images:
    with open(filename, 'rb') as f:
        client.products.upload_image(
            'product-123',
            file=f,
            alt=alt
        )

# Async upload
async with aiofiles.open('product-image.jpg', 'rb') as f:
    content = await f.read()
    image = await async_client.products.upload_image(
        'product-123',
        file=content,
        alt='Product image'
    )

Type Hints & Validation

Pydantic Models

from unified_commerce.models import Product, Order, Customer
from pydantic import BaseModel, validator

class CustomProduct(Product):
    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Price must be positive')
        return v

# Use custom model
product_data = {
    'name': 'Test Product',
    'price': -10  # Will raise ValidationError
}

try:
    product = CustomProduct(**product_data)
except ValueError as e:
    print(f"Validation error: {e}")

Type Checking

from typing import List, Optional
from unified_commerce import UnifiedCommerce
from unified_commerce.models import Product, Order

def process_products(client: UnifiedCommerce) -> List[Product]:
    products: List[Product] = client.products.list(limit=10)
    return products

def get_customer_orders(
    client: UnifiedCommerce,
    customer_id: str
) -> Optional[List[Order]]:
    try:
        orders: List[Order] = client.orders.list(customer_id=customer_id)
        return orders
    except NotFoundError:
        return None

Testing

Mock Client

from unittest.mock import Mock, patch
from unified_commerce import UnifiedCommerce
from unified_commerce.models import Product

def test_fetch_products():
    # Create mock client
    mock_client = Mock(spec=UnifiedCommerce)
    mock_client.products.list.return_value = [
        Product(id='1', name='Test Product', price=99.99)
    ]

    # Use in test
    products = mock_client.products.list(limit=10)
    assert len(products) == 1
    assert products[0].name == 'Test Product'

# Using patch
@patch('unified_commerce.UnifiedCommerce')
def test_with_patch(mock_commerce):
    mock_commerce.return_value.products.list.return_value = [
        Product(id='1', name='Test Product', price=99.99)
    ]

    client = UnifiedCommerce(api_key='test')
    products = client.products.list()

    assert len(products) == 1

Test Utilities

from unified_commerce.testing import (
    mock_product,
    mock_order,
    mock_customer,
    MockUnifiedCommerce
)

# Create mock objects
test_product = mock_product(
    name='Test Product',
    price=99.99
)

test_order = mock_order(
    items=[{'product': test_product, 'quantity': 2}]
)

# Use mock client
mock_client = MockUnifiedCommerce(api_key='test')
mock_client.products.add_mock_response('list', [test_product])

products = mock_client.products.list()
assert products == [test_product]

Logging

Enable Debug Logging

import logging
from unified_commerce import UnifiedCommerce

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Enable SDK debug mode
client = UnifiedCommerce(
    api_key=os.environ['UNIFIED_COMMERCE_API_KEY'],
    debug=True
)

# Or set logging level for SDK
logging.getLogger('unified_commerce').setLevel(logging.DEBUG)

Custom Logger

from unified_commerce import UnifiedCommerce
import logging

# Create custom logger
logger = logging.getLogger('my_app.commerce')

client = UnifiedCommerce(
    api_key=os.environ['UNIFIED_COMMERCE_API_KEY'],
    logger=logger
)

Changelog

Version 1.8.3 (Latest)

Added Python 3.12 support
Improved async performance with httpx 0.25
New Django 5.0 integration helpers
Fixed memory leak in long-running async operations
Added Pydantic v2 support

Version 1.8.0

Breaking: Dropped Python 3.7 support
New async client with httpx backend
Added FastAPI integration utilities
Improved type hints with generics
Added webhook verification helpers

Version 1.7.0

Added batch operations support
New GraphQL query builder
Improved error messages
Added file upload support
Performance optimizations

View full changelog →

Resources

Support

Need help? We're here for you: