klime 1.0.3

Klime SDK for Python

klime

Klime SDK for Python.

Installation

pip install klime

Quick Start

from klime import KlimeClient

client = KlimeClient(
    write_key='your-write-key'
)

# Identify a user
client.identify('user_123', {
    'email': '[e-mail]',
    'name': 'Stefan'
})

# Track an event
client.track('Button Clicked', {
    'button_name': 'Sign up',
    'plan': 'pro'
}, user_id='user_123')

# Associate user with a group and set group traits
client.group('org_456', {
    'name': 'Acme Inc',
    'plan': 'enterprise'
}, user_id='user_123')

# Or just link the user to a group (if traits are already set)
client.group('org_456', user_id='user_123')

# Shutdown gracefully
client.shutdown()

API Reference

Constructor

KlimeClient(
    write_key: str,                    # Required: Your Klime write key
    endpoint: Optional[str] = None,    # Optional: API endpoint (default: https://i.klime.com)
    flush_interval: Optional[int] = None,      # Optional: Milliseconds between flushes (default: 2000)
    max_batch_size: Optional[int] = None,     # Optional: Max events per batch (default: 20, max: 100)
    max_queue_size: Optional[int] = None,      # Optional: Max queued events (default: 1000)
    retry_max_attempts: Optional[int] = None, # Optional: Max retry attempts (default: 5)
    retry_initial_delay: Optional[int] = None, # Optional: Initial retry delay in ms (default: 1000)
    flush_on_shutdown: Optional[bool] = None   # Optional: Auto-flush on exit (default: True)
)

Methods

track(event: str, properties: Optional[Dict] = None, user_id: Optional[str] = None, group_id: Optional[str] = None, ip: Optional[str] = None) -> None

Track a user event. A user_id is required for events to be useful in Klime.

client.track('Button Clicked', {
    'button_name': 'Sign up',
    'plan': 'pro'
}, user_id='user_123')

# With IP address (for geolocation)
client.track('Button Clicked', {
    'button_name': 'Sign up',
    'plan': 'pro'
}, user_id='user_123', ip='192.168.1.1')

Advanced: The group_id parameter is available for multi-tenant scenarios where a user belongs to multiple organizations and you need to specify which organization context the event occurred in.

identify(user_id: str, traits: Optional[Dict] = None, ip: Optional[str] = None) -> None

Identify a user with traits.

client.identify('user_123', {
    'email': '[e-mail]',
    'name': 'Stefan'
}, ip='192.168.1.1')

group(group_id: str, traits: Optional[Dict] = None, user_id: Optional[str] = None, ip: Optional[str] = None) -> None

Associate a user with a group and/or set group traits.

# Associate user with a group and set group traits (most common)
client.group('org_456', {
    'name': 'Acme Inc',
    'plan': 'enterprise'
}, user_id='user_123')

# Just link a user to a group (traits already set or not needed)
client.group('org_456', user_id='user_123')

# Just update group traits (e.g., from a webhook or background job)
client.group('org_456', {
    'plan': 'enterprise',
    'employee_count': 50
})

flush() -> None

Manually flush queued events immediately.

client.flush()

shutdown() -> None

Gracefully shutdown the client, flushing remaining events.

client.shutdown()

Features

• Automatic Batching: Events are automatically batched and sent every 2 seconds or when the batch size reaches 20 events
• Automatic Retries: Failed requests are automatically retried with exponential backoff
• Thread-Safe: Safe to use from multiple threads
• Process Exit Handling: Automatically flushes events on process exit (via atexit)
• Zero Dependencies: Uses only Python standard library

Configuration

Default Values

• flush_interval: 2000ms
• max_batch_size: 20 events
• max_queue_size: 1000 events
• retry_max_attempts: 5 attempts
• retry_initial_delay: 1000ms
• flush_on_shutdown: True

Error Handling

The SDK automatically handles:

• Transient errors (429, 503, network failures): Retries with exponential backoff
• Permanent errors (400, 401): Logs error and drops event
• Rate limiting: Respects Retry-After header

Size Limits

• Maximum event size: 200KB
• Maximum batch size: 10MB
• Maximum events per batch: 100

Events exceeding these limits are rejected and logged.

Flask Example

from flask import Flask, request
from klime import KlimeClient

app = Flask(__name__)
client = KlimeClient(write_key='your-write-key')

@app.route('/api/button-clicked', methods=['POST'])
def button_clicked():
    client.track('Button Clicked', {
        'button_name': request.json.get('buttonName')
    }, user_id=request.json.get('userId'), ip=request.remote_addr)

    return {'success': True}

# Graceful shutdown
import atexit
atexit.register(client.shutdown)

Django Example

from django.http import JsonResponse
from django.views import View
from klime import KlimeClient

client = KlimeClient(write_key='your-write-key')

class ButtonClickView(View):
    def post(self, request):
        client.track('Button Clicked', {
            'button_name': request.POST.get('buttonName')
        }, user_id=str(request.user.id), ip=request.META.get('REMOTE_ADDR'))

        return JsonResponse({'success': True})

Requirements

• Python 3.9 or higher
• No external dependencies (uses only standard library)

License

MIT