Python SDK

Installation

pip install revkeen

Requirements

• Python 3.8+
• httpx for HTTP requests
• pydantic for data validation

Quick Start

from revkeen import RevKeen
import os

client = RevKeen(api_key=os.environ["REVKEEN_API_KEY"])

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

print(customer.data.id)

Async Usage

from revkeen import AsyncRevKeen
import asyncio

async def main():
    client = AsyncRevKeen(api_key=os.environ["REVKEEN_API_KEY"])
    
    customer = await client.customers.create(
        email="john@example.com",
        name="John Doe"
    )
    
    print(customer.data.id)

asyncio.run(main())

Configuration

from revkeen import RevKeen

client = RevKeen(
    # Required
    api_key="rk_live_xxxxxxxx",
    
    # Optional
    base_url="https://api.revkeen.com",  # API base URL
    timeout=30.0,                         # Request timeout in seconds
    max_retries=3,                        # Retry attempts for failures
)

Environment Variables

The SDK automatically reads from environment variables:

export REVKEEN_API_KEY=rk_live_xxxxxxxx

Resources

# List customers
customers = client.customers.list(limit=20, page=1)

# Create a customer
customer = client.customers.create(
    email="john@example.com",
    name="John Doe",
    company_name="Acme Inc",
    metadata={"salesforce_id": "00Q123"}
)

# Retrieve a customer
customer = client.customers.retrieve("cus_xxxxxxxx")

# Update a customer
customer = client.customers.update(
    "cus_xxxxxxxx",
    name="John Smith"
)

# Delete a customer
client.customers.delete("cus_xxxxxxxx")

# List products
products = client.products.list(kind="subscription", is_active=True)

# Create a product
product = client.products.create(
    product_id="pro_monthly",
    name="Pro Plan",
    kind="subscription",
    pricing_model="recurring",
    amount_minor=2900,
    currency="USD",
    interval="month",
    interval_count=1,
    trial_days=14
)

# Retrieve a product
product = client.products.retrieve("prod_xxxxxxxx")

# Update a product
product = client.products.update(
    "prod_xxxxxxxx",
    name="Pro Plan (Updated)"
)

# List subscriptions
subscriptions = client.subscriptions.list(
    customer_id="cus_xxxxxxxx",
    status="active"
)

# Create a subscription
subscription = client.subscriptions.create(
    customer_id="cus_xxxxxxxx",
    product_id="prod_xxxxxxxx",
    trial_days=14
)

# Retrieve a subscription
subscription = client.subscriptions.retrieve("sub_xxxxxxxx")

# Update a subscription (change plan)
subscription = client.subscriptions.update(
    "sub_xxxxxxxx",
    product_id="prod_yyyyyyyy",
    proration="create_prorations"
)

# Cancel a subscription
client.subscriptions.cancel(
    "sub_xxxxxxxx",
    cancel_at_period_end=True
)

# Pause a subscription
client.subscriptions.pause(
    "sub_xxxxxxxx",
    resume_at="2025-03-01T00:00:00Z"
)

# Resume a subscription
client.subscriptions.resume("sub_xxxxxxxx")

# List invoices
invoices = client.invoices.list(
    customer_id="cus_xxxxxxxx",
    status="sent"
)

# Create an invoice
invoice = client.invoices.create(
    customer_id="cus_xxxxxxxx",
    currency="USD",
    due_date="2025-02-15T00:00:00Z",
    line_items=[
        {
            "description": "Consulting Services",
            "quantity": 10,
            "unit_amount_minor": 15000
        }
    ]
)

# Submit for approval
client.invoices.submit("inv_xxxxxxxx")

# Approve an invoice
client.invoices.approve("inv_xxxxxxxx")

# Send an invoice
client.invoices.send("inv_xxxxxxxx")

# Void an invoice
client.invoices.void("inv_xxxxxxxx", reason="Duplicate invoice")

# Download PDF
pdf_bytes = client.invoices.get_pdf("inv_xxxxxxxx")
with open("invoice.pdf", "wb") as f:
    f.write(pdf_bytes)

# Create a checkout session
session = client.checkout_sessions.create(
    mode="subscription",
    product_id="prod_xxxxxxxx",
    customer_id="cus_xxxxxxxx",
    success_url="https://yourapp.com/success",
    cancel_url="https://yourapp.com/cancel"
)

# Get the checkout URL
print(session.data.url)

# Retrieve a session
session = client.checkout_sessions.retrieve("chk_xxxxxxxx")

# Expire a session
client.checkout_sessions.expire("chk_xxxxxxxx")

# List webhook endpoints
webhooks = client.webhooks.list()

# Create a webhook endpoint
webhook = client.webhooks.create(
    url="https://yourapp.com/webhooks/revkeen",
    events=["invoice.paid", "subscription.created"]
)

# Store the secret for verification
print(webhook.data.secret)

# Update a webhook
webhook = client.webhooks.update(
    "whk_xxxxxxxx",
    events=["invoice.paid", "subscription.created", "subscription.cancelled"]
)

# Delete a webhook
client.webhooks.delete("whk_xxxxxxxx")

Pagination

Manual Pagination

page1 = client.customers.list(limit=20, page=1)
page2 = client.customers.list(limit=20, page=2)

print(f"Total: {page1.pagination.total}")
print(f"Pages: {page1.pagination.total_pages}")

Auto-Pagination (Sync)

# Iterate through all results
for customer in client.customers.list():
    print(customer.id, customer.email)

# Get all results as list
all_customers = list(client.customers.list())

Auto-Pagination (Async)

# Iterate through all results
async for customer in client.customers.list():
    print(customer.id, customer.email)

# Get all results as list
all_customers = [c async for c in client.customers.list()]

Error Handling

from revkeen import RevKeen
from revkeen.errors import (
    RevKeenError,
    NotFoundError,
    ValidationError,
    UnauthorizedError,
    ForbiddenError,
    RateLimitError,
    InternalServerError,
)

try:
    customer = client.customers.retrieve("cus_invalid")
except NotFoundError:
    # 404 - Resource not found
    print("Customer not found")
except ValidationError as e:
    # 400/422 - Validation failed
    print(f"Validation error: {e.message}")
    print(f"Details: {e.details}")
except UnauthorizedError:
    # 401 - Invalid API key
    print("Invalid API key")
except ForbiddenError:
    # 403 - Insufficient permissions
    print("Permission denied")
except RateLimitError as e:
    # 429 - Rate limited
    print(f"Rate limited, retry after: {e.retry_after} seconds")
except InternalServerError:
    # 5xx - Server error
    print("Server error, please retry")
except RevKeenError as e:
    # Generic API error
    print(f"API error: {e.code} - {e.message}")

Webhook Verification

FastAPI Example

from revkeen.webhooks import verify_webhook, WebhookVerificationError
from fastapi import FastAPI, Request, HTTPException
import os

app = FastAPI()

@app.post("/webhooks/revkeen")
async def webhook_handler(request: Request):
    payload = await request.body()
    signature = request.headers.get("x-revkeen-signature")
    
    try:
        event = verify_webhook(
            payload.decode(),
            signature,
            os.environ["WEBHOOK_SECRET"]
        )
    except WebhookVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    if event.type == "invoice.paid":
        await handle_invoice_paid(event.data)
    elif event.type == "subscription.created":
        await handle_subscription_created(event.data)
    elif event.type == "subscription.cancelled":
        await handle_subscription_cancelled(event.data)
    
    return {"received": True}

Flask Example

from flask import Flask, request, jsonify
from revkeen.webhooks import verify_webhook, WebhookVerificationError

app = Flask(__name__)

@app.route("/webhooks/revkeen", methods=["POST"])
def webhook_handler():
    payload = request.get_data(as_text=True)
    signature = request.headers.get("x-revkeen-signature")
    
    try:
        event = verify_webhook(payload, signature, WEBHOOK_SECRET)
    except WebhookVerificationError:
        return "Invalid signature", 400
    
    if event.type == "invoice.paid":
        handle_invoice_paid(event.data)
    
    return jsonify({"received": True})

Type Hints

The SDK includes comprehensive type hints:

from revkeen import RevKeen
from revkeen.types import (
    # Resource types
    Customer,
    Product,
    Subscription,
    Invoice,
    CheckoutSession,
    
    # Request types
    CustomerCreateParams,
    ProductCreateParams,
    SubscriptionCreateParams,
    
    # Enums
    SubscriptionStatus,
    InvoiceStatus,
    ProductKind,
    BillingInterval,
)

def create_customer(params: CustomerCreateParams) -> Customer:
    response = client.customers.create(**params)
    return response.data

Request Options

Pass request options to individual calls:

# Custom timeout for slow operations
customers = client.customers.list(
    _request_timeout=60.0
)

# Idempotency key for POST requests
customer = client.customers.create(
    email="john@example.com",
    _idempotency_key="unique-request-id"
)

# Custom headers
customer = client.customers.retrieve(
    "cus_xxx",
    _headers={"X-Custom-Header": "value"}
)

Debugging

Enable debug logging:

import logging

logging.basicConfig(level=logging.DEBUG)

# Or just for RevKeen
logging.getLogger("revkeen").setLevel(logging.DEBUG)

Framework Examples

Django

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from revkeen import RevKeen
from revkeen.webhooks import verify_webhook
import json

client = RevKeen()

@csrf_exempt
@require_POST
def create_customer(request):
    data = json.loads(request.body)
    customer = client.customers.create(
        email=data["email"],
        name=data.get("name")
    )
    return JsonResponse(customer.data.dict())

@csrf_exempt
@require_POST
def webhook(request):
    payload = request.body.decode()
    signature = request.headers.get("x-revkeen-signature")
    
    event = verify_webhook(payload, signature, settings.WEBHOOK_SECRET)
    
    if event.type == "invoice.paid":
        handle_invoice_paid(event.data)
    
    return JsonResponse({"received": True})

FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from revkeen import AsyncRevKeen

app = FastAPI()
client = AsyncRevKeen()

class CustomerCreate(BaseModel):
    email: str
    name: str | None = None

@app.post("/api/customers")
async def create_customer(data: CustomerCreate):
    customer = await client.customers.create(
        email=data.email,
        name=data.name
    )
    return customer.data

Related Resources