> ## Documentation Index
> Fetch the complete documentation index at: https://docs.snappay.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Complete guide for integrating SnapPay with Python applications using async patterns

## 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

```bash theme={null}
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):**

```bash theme={null}
export SNAPPAY_API_KEY="pk_test_xxxxxxxxxx"
```

**Direct Configuration:**

```python theme={null}
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="user@example.com")

    # Using direct API key
    async with SnapPay(api_key="pk_test_xxxxxxxxxx") as client:
        customer = await client.customers.get("cus_123", email="user@example.com")

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).

```python theme={null}
from snappay import SnapPay

async with SnapPay() as client:
    customer = await client.customers.get(
        customer_id="cus_123",
        email="user@example.com", # 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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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](/events/index)

### Stream Events (Async Generator)

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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="test@example.com")

        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

```python theme={null}
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)
```