Python

A Python client for the LightRate token management API, providing easy-to-use methods for consuming tokens with local token bucket support for improved performance.

Installation

Install the package using pip:


pip install lightrate-client

Or with pipenv:


pipenv install lightrate-client

Configuration

Configure the client with your API credentials:


from lightrate_client import LightRateClient
 
# Simple usage - pass your API key and application ID
client = LightRateClient('your_api_key', 'your_application_id')
 
# With additional options
client = LightRateClient(
    'your_api_key',
    'your_application_id',
    timeout=30,  # optional, defaults to 30 seconds
    retry_attempts=3,  # optional, defaults to 3
    default_local_bucket_size=10  # optional, defaults to 5
)

Basic Usage


from lightrate_client import LightRateClient
 
# Create a client with your API key and application ID
client = LightRateClient('your_api_key', 'your_application_id')
 
# With additional options
client = LightRateClient(
    'your_api_key',
    'your_application_id',
    timeout=60,
    default_local_bucket_size=20
)

Consuming Tokens

Basic Token Consumption (Local Bucket)

The Python client uses local token buckets for efficient rate limiting. Local buckets automatically refill from the server when needed.


# Consume tokens by operation (uses local token bucket)
response = client.consume_local_bucket_token(
    user_identifier='user123',
    operation='send_email'
)
 
# Or consume tokens by path
response = client.consume_local_bucket_token(
    user_identifier='user123',
    path='/api/v1/emails/send',
    http_method='POST'
)
 
if response.success:
    print(f"Token consumed from {'local bucket' if response.used_local_token else 'server'}")
else:
    print("Rate limited: No tokens available")

Using Local Token Buckets

The client supports local token buckets for improved performance:


# Configure client with bucket sizes
client = LightRateClient(
    'your_api_key',
    'your_application_id',
    default_local_bucket_size=20
)
 
# Consume tokens using local bucket (more efficient)
result = client.consume_local_bucket_token(
    user_identifier='user123',
    operation='send_email'
)
 
if result.success:
    print(f"Success! Used {'local' if result.used_local_token else 'server'} token")
    print(f"Bucket status: {result.bucket_status}")
else:
    print('Rate limited: No tokens available')

Direct HTTP API Usage

For advanced use cases where you need to bypass the local bucket and call the HTTP API directly:


# Direct API call (bypasses local bucket)
response = client.consume_tokens(
    user_identifier='user123',
    tokens_requested=1,
    operation='send_email'
)
 
if response.tokens_consumed > 0:
    print(f"Consumed {response.tokens_consumed} tokens. Remaining: {response.tokens_remaining}")
else:
    print("Rate limited: No tokens available")

Complete Example


from lightrate_client import LightRateClient
 
# Create a client with your API key and application ID
client = LightRateClient('your_api_key', 'your_application_id')
 
try:
    # Consume tokens using local bucket
    response = client.consume_local_bucket_token(
        user_identifier='user123',
        operation='send_email'
    )
 
    if response.success:
        print(f"Token consumed from {'local bucket' if response.used_local_token else 'server'}")
        # Proceed with your operation
        send_email_to_user('user123')
    else:
        print('Rate limit exceeded. Please try again later.')
        # Handle rate limiting
 
except LightRateClient.UnauthorizedError as e:
    print(f"Authentication failed: {e.message}")
except LightRateClient.TooManyRequestsError as e:
    print(f"Rate limited: {e.message}")
except LightRateClient.APIError as e:
    print(f"API Error ({e.status_code}): {e.message}")
except LightRateClient.NetworkError as e:
    print(f"Network error: {e.message}")

Error Handling

The client provides comprehensive error handling with specific exception types:


from lightrate_client import LightRateClient
 
try:
    response = client.consume_tokens(
        user_identifier='user123',
        tokens_requested=1,
        operation='send_email'
    )
except LightRateClient.UnauthorizedError as e:
    print(f"Authentication failed: {e.message}")
except LightRateClient.NotFoundError as e:
    print(f"Resource not found: {e.message}")
except LightRateClient.APIError as e:
    print(f"API Error ({e.status_code}): {e.message}")
except LightRateClient.NetworkError as e:
    print(f"Network error: {e.message}")
except LightRateClient.TimeoutError as e:
    print(f"Request timed out: {e.message}")

Available Exception Types

LightRateClient.LightRateError - Base error class
LightRateClient.ConfigurationError - Configuration-related errors
LightRateClient.AuthenticationError - Authentication-related errors
LightRateClient.APIError - Base API error class
LightRateClient.BadRequestError - 400 errors
LightRateClient.UnauthorizedError - 401 errors
LightRateClient.ForbiddenError - 403 errors
LightRateClient.NotFoundError - 404 errors
LightRateClient.UnprocessableEntityError - 422 errors
LightRateClient.TooManyRequestsError - 429 errors
LightRateClient.InternalServerError - 500 errors
LightRateClient.ServiceUnavailableError - 503 errors
LightRateClient.NetworkError - Network-related errors
LightRateClient.TimeoutError - Request timeout errors

API Reference

Classes

`LightRateClient`

Main client class for interacting with the LightRate API.

Constructor:


LightRateClient(api_key: str, application_id: str, **options)

Parameters:

api_key (str, required): Your LightRate API key
application_id (str, required): Your LightRate Application ID
timeout (int, optional): Request timeout in seconds (default: 30)
retry_attempts (int, optional): Number of retry attempts (default: 3)
default_local_bucket_size (int, optional): Default local bucket size (default: 5)

Methods:

consume_tokens(user_identifier, tokens_requested, operation=None, path=None, http_method=None) -> ConsumeTokensResponse
consume_local_bucket_token(user_identifier, operation=None, path=None, http_method=None) -> ConsumeLocalBucketTokenResponse
get_all_bucket_statuses() -> dict
reset_all_buckets() -> None
get_configuration() -> dict