Skip to main content

Python SDK

The Azotte Python SDK provides a comprehensive interface for integrating subscription billing and payment processing into Python applications.

Installation​

Using pip​

pip install azotte-python

Using conda​

conda install -c azotte azotte-python

From source​

git clone https://github.com/azotte/azotte-python.git
cd azotte-python
pip install -e .

Quick Start​

Basic Setup​

import azotte
from azotte import Customer, Subscription, Plan

# Initialize the client
client = azotte.Client(
    api_key='sk_live_...',  # Your secret key
    tenant_id='tenant_123',  # Optional: multi-tenant setup
    environment='production'  # or 'sandbox'
)

# Create a customer
customer = client.customers.create(
    name='John Doe',
    email='john@example.com',
    phone='+1234567890'
)

# Create a subscription
subscription = client.subscriptions.create(
    customer_id=customer.id,
    plan_id='plan_basic_monthly',
    payment_method='pm_card_visa'
)

print(f"Subscription created: {subscription.id}")

Configuration​

Environment Variables​

import os
from azotte import Client

# Load from environment variables
client = Client(
    api_key=os.getenv('AZOTTE_API_KEY'),
    tenant_id=os.getenv('AZOTTE_TENANT_ID'),
    webhook_secret=os.getenv('AZOTTE_WEBHOOK_SECRET')
)

Advanced Configuration​

from azotte import Client, Config

config = Config(
    api_key='sk_live_...',
    base_url='https://api.azotte.com',  # Custom endpoint
    timeout=30,  # Request timeout in seconds
    retries=3,   # Number of retry attempts
    backoff_factor=2,  # Exponential backoff factor
    verify_ssl=True,   # SSL certificate verification
    user_agent='MyApp/1.0'  # Custom user agent
)

client = Client(config)

Customer Management​

Create Customer​

from azotte.models import Customer

customer = client.customers.create(
    name='Jane Smith',
    email='jane@example.com',
    phone='+1987654321',
    metadata={
        'source': 'website',
        'campaign': 'summer_2024'
    },
    address={
        'line1': '123 Main St',
        'city': 'San Francisco',
        'state': 'CA',
        'postal_code': '94105',
        'country': 'US'
    }
)

Retrieve Customer​

# Get customer by ID
customer = client.customers.retrieve('cus_123')

# Get customer with expanded data
customer = client.customers.retrieve(
    'cus_123',
    expand=['subscriptions', 'payment_methods']
)

print(f"Customer: {customer.name}")
print(f"Subscriptions: {len(customer.subscriptions)}")

Update Customer​

customer = client.customers.update(
    'cus_123',
    name='Jane Doe',
    email='jane.doe@example.com',
    metadata={'updated_at': '2024-01-15'}
)

List Customers​

# Basic listing
customers = client.customers.list(limit=10)

# With filters
customers = client.customers.list(
    created_after='2024-01-01',
    email_contains='@company.com',
    has_subscription=True,
    limit=50
)

# Pagination
for customer in client.customers.auto_paging_iter():
    print(f"Processing customer: {customer.id}")

Subscription Management​

Create Subscription​

from azotte.models import Subscription
from datetime import datetime, timedelta

subscription = client.subscriptions.create(
    customer_id='cus_123',
    plan_id='plan_premium_annual',
    trial_period_days=14,
    start_date=datetime.now() + timedelta(days=7),
    payment_method='pm_card_123',
    metadata={'promo_code': 'WELCOME20'}
)

Subscription with Custom Pricing​

subscription = client.subscriptions.create(
    customer_id='cus_123',
    items=[
        {
            'plan_id': 'plan_basic',
            'quantity': 1
        },
        {
            'plan_id': 'addon_analytics',
            'quantity': 1
        }
    ],
    discount_coupon='SAVE10',
    billing_cycle_anchor='now'
)

Update Subscription​

# Change plan
subscription = client.subscriptions.update(
    'sub_123',
    plan_id='plan_enterprise',
    proration_behavior='create_prorations'
)

# Pause subscription
subscription = client.subscriptions.pause('sub_123')

# Resume subscription 
subscription = client.subscriptions.resume('sub_123')

# Cancel subscription
subscription = client.subscriptions.cancel(
    'sub_123',
    at_period_end=True,
    cancellation_reason='customer_request'
)

Payment Processing​

Payment Methods​

# Add payment method
payment_method = client.payment_methods.create(
    customer_id='cus_123',
    type='card',
    card={
        'number': '4111111111111111',
        'exp_month': 12,
        'exp_year': 2025,
        'cvc': '123'
    }
)

# Set default payment method
client.customers.update(
    'cus_123',
    default_payment_method=payment_method.id
)

One-time Payments​

from azotte.models import Payment

payment = client.payments.create(
    amount=2999,  # $29.99
    currency='usd',
    customer_id='cus_123',
    payment_method='pm_123',
    description='One-time setup fee',
    confirm=True
)

if payment.status == 'succeeded':
    print(f"Payment successful: {payment.id}")
else:
    print(f"Payment failed: {payment.failure_reason}")

Refunds​

# Full refund
refund = client.refunds.create(
    payment_id='pay_123',
    reason='requested_by_customer'
)

# Partial refund
refund = client.refunds.create(
    payment_id='pay_123',
    amount=1500,  # $15.00
    reason='defective_product'
)

Webhook Handling​

Flask Integration​

from flask import Flask, request, jsonify
import azotte

app = Flask(__name__)
client = azotte.Client(api_key='sk_live_...')

@app.route('/webhooks/azotte', methods=['POST'])
def handle_webhook():
    payload = request.get_data()
    signature = request.headers.get('azotte-signature')
    
    try:
        event = client.webhooks.construct_event(
            payload, signature, 'whsec_...'
        )
    except azotte.SignatureVerificationError:
        return jsonify({'error': 'Invalid signature'}), 400
    
    # Handle the event
    if event.type == 'subscription.created':
        handle_subscription_created(event.data)
    elif event.type == 'payment.succeeded':
        handle_payment_succeeded(event.data)
    elif event.type == 'payment.failed':
        handle_payment_failed(event.data)
    
    return jsonify({'status': 'success'})

def handle_subscription_created(subscription):
    print(f"New subscription: {subscription.id}")
    # Send welcome email, grant access, etc.

def handle_payment_succeeded(payment):
    print(f"Payment received: ${payment.amount / 100}")
    # Update user account, send receipt, etc.

def handle_payment_failed(payment):
    print(f"Payment failed: {payment.failure_reason}")
    # Send dunning email, suspend access, etc.

Django Integration​

# views.py
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
import azotte
import json

client = azotte.Client(api_key=settings.AZOTTE_API_KEY)

@csrf_exempt
@require_POST
def azotte_webhook(request):
    payload = request.body
    signature = request.META.get('HTTP_AZOTTE_SIGNATURE')
    
    try:
        event = client.webhooks.construct_event(
            payload, signature, settings.AZOTTE_WEBHOOK_SECRET
        )
    except azotte.SignatureVerificationError:
        return HttpResponse(status=400)
    
    # Process event asynchronously with Celery
    process_azotte_event.delay(event.to_dict())
    
    return HttpResponse(status=200)

# tasks.py (Celery)
from celery import shared_task

@shared_task
def process_azotte_event(event_data):
    event_type = event_data['type']
    data = event_data['data']
    
    if event_type == 'subscription.updated':
        # Update local subscription record
        update_subscription_status(data['id'], data['status'])

Advanced Features​

Metered Billing​

# Report usage
usage_record = client.subscription_items.create_usage_record(
    subscription_item_id='si_123',
    quantity=100,
    timestamp=datetime.now(),
    action='increment'  # or 'set'
)

# Get usage summary
usage_summary = client.subscription_items.list_usage_records(
    subscription_item_id='si_123',
    start_date='2024-01-01',
    end_date='2024-01-31'
)

Invoicing​

# Create invoice
invoice = client.invoices.create(
    customer_id='cus_123',
    auto_advance=True,
    collection_method='charge_automatically'
)

# Add invoice items
client.invoice_items.create(
    customer_id='cus_123',
    amount=2500,
    currency='usd',
    description='Professional services'
)

# Finalize and send
invoice = client.invoices.finalize('in_123')
client.invoices.send(invoice.id)

Disputes and Chargebacks​

# List disputes
disputes = client.disputes.list(status='needs_response')

# Submit evidence
client.disputes.submit_evidence(
    'dp_123',
    evidence={
        'customer_communication': 'Email thread with customer',
        'receipt': 'Payment receipt #12345',
        'shipping_documentation': 'Tracking number: 1Z999AA1...'
    }
)

Error Handling​

Exception Handling​

import azotte
from azotte.errors import (
    AzotteError,
    APIError,
    AuthenticationError,
    ValidationError,
    RateLimitError
)

try:
    customer = client.customers.create(
        name='',  # Invalid: empty name
        email='invalid-email'  # Invalid: bad format
    )
except ValidationError as e:
    print(f"Validation failed: {e.user_message}")
    for field, errors in e.field_errors.items():
        print(f"  {field}: {', '.join(errors)}")

except AuthenticationError as e:
    print(f"Authentication failed: {e}")
    # Handle invalid API key

except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")
    # Implement exponential backoff

except APIError as e:
    print(f"API error: {e.message}")
    print(f"Request ID: {e.request_id}")
    print(f"Status code: {e.status_code}")

except AzotteError as e:
    print(f"Azotte error: {e}")
    # Handle other Azotte-specific errors

except Exception as e:
    print(f"Unexpected error: {e}")
    # Handle network errors, etc.

Retry Logic​

import time
from azotte.errors import RateLimitError, APIError

def create_customer_with_retry(customer_data, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            return client.customers.create(**customer_data)
        except RateLimitError as e:
            if attempt == max_retries:
                raise
            time.sleep(e.retry_after or 2 ** attempt)
        except APIError as e:
            if e.status_code >= 500 and attempt < max_retries:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise

Testing​

Mock Client​

import azotte
from unittest.mock import Mock, patch

# Mock the entire client
mock_client = Mock(spec=azotte.Client)
mock_client.customers.create.return_value = Mock(
    id='cus_test_123',
    name='Test Customer',
    email='test@example.com'
)

# Test your function
def test_create_customer():
    with patch('azotte.Client', return_value=mock_client):
        customer = create_customer('Test Customer', 'test@example.com')
        assert customer.id == 'cus_test_123'

Test Mode​

# Use test API keys for testing
test_client = azotte.Client(
    api_key='sk_test_...',
    environment='sandbox'
)

# Create test data
test_customer = test_client.customers.create(
    name='Test Customer',
    email='test@example.com'
)

# Test payments with test card numbers
test_payment = test_client.payments.create(
    amount=2000,
    currency='usd',
    customer_id=test_customer.id,
    payment_method_data={
        'type': 'card',
        'card': {
            'number': '4242424242424242',  # Test Visa card
            'exp_month': 12,
            'exp_year': 2025,
            'cvc': '123'
        }
    }
)

Performance Optimization​

Connection Pooling​

import azotte
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

# Custom session with connection pooling
session = requests.Session()
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    backoff_factor=1
)
adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=retry_strategy
)
session.mount("https://", adapter)

client = azotte.Client(
    api_key='sk_live_...',
    session=session
)

Async Support​

import asyncio
import azotte

# Async client
async_client = azotte.AsyncClient(api_key='sk_live_...')

async def create_multiple_customers(customer_data_list):
    tasks = [
        async_client.customers.create(**data)
        for data in customer_data_list
    ]
    customers = await asyncio.gather(*tasks)
    return customers

# Usage
customer_data = [
    {'name': 'Customer 1', 'email': 'c1@example.com'},
    {'name': 'Customer 2', 'email': 'c2@example.com'},
]

customers = asyncio.run(create_multiple_customers(customer_data))

Best Practices​

Idempotency​

import uuid

# Use idempotency keys for critical operations
idempotency_key = str(uuid.uuid4())

customer = client.customers.create(
    name='John Doe',
    email='john@example.com',
    idempotency_key=idempotency_key
)

Pagination​

# Efficient pagination for large datasets
def process_all_customers():
    for customer in client.customers.auto_paging_iter(limit=100):
        process_customer(customer)
        
def process_customer(customer):
    print(f"Processing {customer.name}")
    # Your business logic here

Next Steps​