Skip to main content

​
Overview

The SnapPay Python SDK provides a comprehensive, async-first interface for integrating payment processing, subscription management, and real-time event streaming into your Python applications. Built with modern asyncio patterns and type safety, it offers seamless integration with SnapPay’s platform. Requirements: Python 3.8+

​
Key Features

  • Async-first design with aiohttp for optimal performance
  • Context manager support for automatic session management
  • Real-time event streaming via Server-Sent Events (SSE)
  • Full type safety with TypedDict definitions
  • Service-based architecture for organized functionality
  • Automatic retry logic and error handling
  • Environment-based configuration

​
Installation

pip install snappay

​
Configuration & Initialization

The SDK is configured through environment variables and the client’s constructor. All operations require proper authentication and should use the async context manager pattern for optimal resource management.

​
API Key Authentication

Authentication is handled via an API key that must start with pk_test_ (for testing) or pk_live_ (for production). Configure your API key in one of these ways: Environment Variable (Recommended): 
export SNAPPAY_API_KEY="pk_test_xxxxxxxxxx"
Direct Configuration: 
import asyncio
from snappay import SnapPay

async def main():
    # Using environment variable (recommended)
    async with SnapPay() as client:
        customer = await client.customers.get("cus_123", email="[email protected]")

    # Using direct API key
    async with SnapPay(api_key="pk_test_xxxxxxxxxx") as client:
        customer = await client.customers.get("cus_123", email="[email protected]")

asyncio.run(main())

​
Core Methods

All methods are async coroutines and must be called with await. The SDK uses a service-based architecture where methods are organized into logical groups: customers, checkout, access, and usage.

​
Customer Management

​
customers.get

Retrieves or creates a customer record (upsert logic). 
from snappay import SnapPay

async with SnapPay() as client:
    customer = await client.customers.get(
        customer_id="cus_123",
        email="[email protected]", # Optional
        name="John Doe"  # Optional
    )
    print(f"Customer ID: {customer['customer_id']}")
Parameters:
  • customer_id (str): Customer identifier
  • email (Optional[str]): Customer email address
  • name (Optional[str]): Customer full name
Returns: Customer object with:
  • customer_id (str): SnapPay customer ID
  • email (str): Customer email
  • name (Optional[str]): Customer name

​
Checkout Sessions

​
checkout.create_session

Creates a payment checkout session URL for customer purchases. 
from snappay import SnapPay, Provider

async with SnapPay() as client:
    session = await client.checkout.create_session(
        customer_id="cus_123",
        product_id="premium-plan",
        success_url="https://yourapp.com/success",
        cancel_url="https://yourapp.com/cancel",  # Optional
        price_id="b33a17d0-0e8a-4762-b9e5-dc19fde35daa" # Optional
    )
    print(f"Checkout URL: {session['url']}")
Parameters:
  • customer_id (str): SnapPay customer ID
  • product_id (str): Product ID from your SnapPay dashboard
  • success_url (str): URL to redirect after successful payment
  • cancel_url (Optional[str]): URL to redirect on cancellation
  • price_id (Optional[str]): The specific price you want to create the checkout session with
Returns: CheckoutSession object with:
  • session_id (str): Unique session identifier
  • url (str): Checkout URL for customer
  • expires_at (str): Session expiration timestamp (additional fields may be present depending on server response)

​
Access Control

​
access.check

Checks if a customer has access to a specific feature based on their subscription and usage limits. 
from snappay import SnapPay

async with SnapPay() as client:
    access = await client.access.check(
        customer_id="cus_123",
        feature_id="premium-features"
    )

    if access["has_access"]:
        allowance = access.get("allowance")
        usage = access.get("usage")
        if allowance is not None and usage is not None:
            print(f"Access granted. {usage}/{allowance} used")
        else:
            print("Access granted (unlimited feature)")
    else:
        print("Access denied. Upgrade required.")
Parameters:
  • customer_id (str): SnapPay customer ID
  • feature_id (str): Feature identifier
Returns: AccessCheck object with:
  • has_access (bool): Whether customer has access
  • feature_id (str): Feature being checked
  • usage (Optional[int]): Total usage (null if the feature is unlimited/no access)
  • allowance (Optional[int]): Maximum usage allowed (null if the feature is unlimited/no access)
  • next_reset_at (Optional[int]): Next reset timestamp (null if the feature is unlimited/no access)

​
Usage Tracking

​
usage.track

Reports usage for a metered feature. This action is idempotent to prevent duplicate tracking. 
from snappay import SnapPay

async with SnapPay() as client:
    result = await client.usage.track(
        customer_id="cus_123",
        feature_id="ai-messages",
        usage=1
    )
    print(f"Usage tracked successfully: {result}")
Parameters:
  • customer_id (str): SnapPay customer ID
  • feature_id (str): Feature identifier for usage tracking
  • usage (int): Usage amount to track
  • idempotency_key (Optional[str]): Prevents duplicate tracking (default to a random UUID if not set)
Returns: TrackUsageResponse object with tracking confirmation details.

​
usage.get

Retrieves current usage details for a customer’s feature. 
from snappay import SnapPay

async with SnapPay() as client:
    usage_list = await client.usage.get(
        customer_id="cus_123",
        feature_id="ai-messages"
    )
    # API may return multiple records; iterate or select first
    for usage in usage_list:
        print(f"Total usage: {usage['total_usage']}")
        if usage.get('remaining') is not None:
            print(f"Remaining: {usage['remaining']}")
        if usage.get('limit') is not None:
            print(f"Limit: {usage['limit']}")
Parameters:
  • customer_id (str): SnapPay customer ID
  • feature_id (str): Feature identifier
Returns: List of GetUsageResponse objects. Each object will contain:
  • total_usage (int): The cumulative amount of the feature used by the customer.
  • product_id (str): Product identifier associated with the usage.
  • feature_id (str): Specific feature identifier for which the usage is tracked.
  • remaining (Optional[int]): The remaining amount of the feature usage allowed.
  • limit (Optional[int]): The maximum usage limit set for the feature.
  • next_reset_at (Optional[str]): Timestamp indicating when the usage will reset.

​
Real-time Event Handling

SnapPay provides real-time events via Server-Sent Events (SSE). Subscribe to standardized backend events without managing webhooks.

​
🚫 What You DON’T Need:

  • No webhook endpoints to create and maintain
  • No webhook signature verification
  • No webhook retry logic or failure handling
  • No webhook security concerns
  • No debugging webhook delivery issues

​
✅ What You GET:

  • Real-time events delivered instantly to your application
  • Guaranteed delivery with automatic reconnection
  • Clean, structured data - no raw webhook payloads
  • Multiple consumption patterns to fit your architecture
  • Type-safe event objects with full IDE support

​
Supported Events

See the list of supported events: Server-sent Event Types

​
Stream Events (Async Generator)

import asyncio
from snappay import SnapPay

async def main():
    async with SnapPay() as client:
        async for event in client.stream_events():
            print(f"Event: {event.type}")
            print(f"ID: {event.id}")
            print(f"Data: {event.data}")

asyncio.run(main())

​
Event Handlers

import asyncio
from snappay import SnapPay, SSEEvent, SSEEventType

def handle_subscription_updated(event: SSEEvent):
    print(f"Subscription updated: {event.data}")

def handle_any(event: SSEEvent):
    print(f"Event: {event.type} -> {event.data}")

async def main():
    async with SnapPay() as client:
        client.on_event(SSEEventType.SUBSCRIPTION_UPDATED, handle_subscription_updated)
        client.on_any_event(handle_any)
        await client.start_events()
        await asyncio.sleep(60)
        await client.stop_events()

asyncio.run(main())

​
Error Handling

The SDK provides a comprehensive exception hierarchy for robust error handling: 
import asyncio
from snappay import (
    SnapPay, SnapPayError, AuthenticationError,
    ValidationError, RateLimitError, NotFoundError,
    ConflictError, PaymentError, ServerError
)

async def comprehensive_error_handling():
    async with SnapPay() as client:
        try:
            await client.customers.get("cus_123", email="[email protected]")

        except AuthenticationError as e:
            print(f"Invalid API key: {e}")

        except ValidationError as e:
            print(f"Invalid parameters: {e}")

        except RateLimitError as e:
            print(f"Rate limit exceeded: {e}")

        except NotFoundError as e:
            print(f"Resource not found: {e}")

        except ConflictError as e:
            print(f"Resource conflict: {e}")

        except PaymentError as e:
            print(f"Payment processing error: {e}")

        except ServerError as e:
            print(f"Server error: {e}")

        except SnapPayError as e:
            print(f"General SnapPay error: {e}")

        except Exception as e:
            print(f"Unexpected error: {e}")

asyncio.run(comprehensive_error_handling())

​
Exception Hierarchy

SnapPayError                    # Base exception
├── AuthenticationError          # Invalid API key or authentication
├── ValidationError              # Invalid parameters or request data
├── RateLimitError              # API rate limit exceeded
├── NotFoundError               # Resource not found
├── ConflictError               # Resource conflict (e.g., duplicate)
├── PaymentError                # Payment processing failures
└── ServerError                 # Server-side errors (5xx responses)