CompactSaaS External API - SDK Code Snippets

Complete code examples for integrating with the CompactSaaS External API in common programming languages.

For API endpoint details, see the API Reference. For a quick introduction, see the Getting Started Guide.

Table of Contents

• JavaScript / Node.js
• Python
• cURL
• Environment Setup
• Error Handling Best Practices
• Retry Logic
• Batch Tracking

JavaScript / Node.js

Installation

# No additional packages required for Node.js 18+ (native fetch)
# For older Node.js versions or additional features:
npm install node-fetch  # For Node.js < 18
npm install axios       # Alternative HTTP client

Environment Setup

# Set your API key as an environment variable
export COMPACTSAAS_API_KEY="pk_live_your_key_here"

# For staging/development
export COMPACTSAAS_API_KEY="pk_test_your_key_here"
export COMPACTSAAS_BASE_URL="https://api.compactsaasstaging.click/v1"

Basic Client (Native Fetch)

/**
 * CompactSaaS API Client
 * Works with Node.js 18+ or modern browsers
 */
const COMPACTSAAS_API_KEY = process.env.COMPACTSAAS_API_KEY;
const BASE_URL = process.env.COMPACTSAAS_BASE_URL || 'https://api.compactsaas.com/v1';

/**
 * Track an analytics event
 * @param {string} event - Event name (e.g., 'page_view', 'button_click')
 * @param {Object} properties - Optional event properties
 * @param {string} timestamp - Optional ISO 8601 timestamp
 * @returns {Promise<{success: boolean}>}
 */
async function trackEvent(event, properties = {}, timestamp = null) {
  const payload = { event, properties };
  if (timestamp) {
    payload.timestamp = timestamp;
  }

  const response = await fetch(`${BASE_URL}/track`, {
    method: 'POST',
    headers: {
      'x-api-key': COMPACTSAAS_API_KEY,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify(payload),
  });

  if (!response.ok) {
    const error = await response.json();
    throw new Error(`API error ${response.status}: ${error.message} (${error.code})`);
  }

  return response.json();
}

/**
 * Check API health status
 * @returns {Promise<{status: string, timestamp: string, version: string}>}
 */
async function healthCheck() {
  const response = await fetch(`${BASE_URL}/health`);

  if (!response.ok) {
    throw new Error(`Health check failed: ${response.status}`);
  }

  return response.json();
}

// Usage examples
async function main() {
  // Track a page view
  await trackEvent('page_view', {
    page: '/pricing',
    referrer: 'https://google.com'
  });
  console.log('Page view tracked');

  // Track a button click
  await trackEvent('button_click', {
    button_id: 'signup_cta',
    button_text: 'Get Started'
  });
  console.log('Button click tracked');

  // Track a purchase with timestamp
  await trackEvent('purchase_completed', {
    product_id: 'prod_123',
    amount: 99.99,
    currency: 'USD',
  }, '2024-01-15T10:30:00Z');
  console.log('Purchase tracked');

  // Check health
  const health = await healthCheck();
  console.log('API Status:', health.status);
}

main().catch(console.error);

TypeScript Types

// types.ts - TypeScript type definitions for CompactSaaS API

/** Request payload for tracking events */
interface TrackEventRequest {
  /** Event name (1-255 characters) */
  event: string;
  /** Optional key-value pairs with event data */
  properties?: Record<string, unknown>;
  /** Optional ISO 8601 timestamp */
  timestamp?: string;
}

/** Response from track endpoint */
interface TrackEventResponse {
  success: boolean;
}

/** Response from health endpoint */
interface HealthResponse {
  status: 'ok';
  timestamp: string;
  version: string;
}

/** Error response from API */
interface ErrorResponse {
  error: string;
  message: string;
  code: 'VALIDATION_ERROR' | 'UNAUTHORIZED' | 'ACCOUNT_DISABLED' |
        'RATE_LIMIT_EXCEEDED' | 'INTERNAL_ERROR' | 'NOT_FOUND';
}

/** Custom error class for API errors */
class CompactSaaSError extends Error {
  constructor(
    public statusCode: number,
    public errorCode: string,
    message: string
  ) {
    super(message);
    this.name = 'CompactSaaSError';
  }
}

Full TypeScript Client with Error Handling

// compactsaas-client.ts - Full-featured TypeScript client

interface CompactSaaSConfig {
  apiKey: string;
  baseUrl?: string;
  timeout?: number;
}

interface TrackOptions {
  event: string;
  properties?: Record<string, unknown>;
  timestamp?: string;
}

class CompactSaaSClient {
  private apiKey: string;
  private baseUrl: string;
  private timeout: number;

  constructor(config: CompactSaaSConfig) {
    if (!config.apiKey) {
      throw new Error('API key is required');
    }
    this.apiKey = config.apiKey;
    this.baseUrl = config.baseUrl || 'https://api.compactsaas.com/v1';
    this.timeout = config.timeout || 10000;
  }

  /**
   * Track an analytics event
   */
  async track(options: TrackOptions): Promise<{ success: boolean }> {
    const { event, properties = {}, timestamp } = options;

    const payload: Record<string, unknown> = { event, properties };
    if (timestamp) {
      payload.timestamp = timestamp;
    }

    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), this.timeout);

    try {
      const response = await fetch(`${this.baseUrl}/track`, {
        method: 'POST',
        headers: {
          'x-api-key': this.apiKey,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify(payload),
        signal: controller.signal,
      });

      clearTimeout(timeoutId);

      if (!response.ok) {
        await this.handleError(response);
      }

      return response.json();
    } catch (error) {
      clearTimeout(timeoutId);
      if (error instanceof Error && error.name === 'AbortError') {
        throw new Error(`Request timed out after ${this.timeout}ms`);
      }
      throw error;
    }
  }

  /**
   * Check API health
   */
  async health(): Promise<{ status: string; timestamp: string; version: string }> {
    const response = await fetch(`${this.baseUrl}/health`);

    if (!response.ok) {
      throw new Error(`Health check failed: ${response.status}`);
    }

    return response.json();
  }

  private async handleError(response: Response): Promise<never> {
    const errorData = await response.json().catch(() => ({}));

    switch (response.status) {
      case 400:
        throw new Error(`Validation error: ${errorData.message || 'Invalid request'}`);
      case 401:
        throw new Error('Invalid or revoked API key');
      case 403:
        throw new Error('Account disabled - please contact support');
      case 429:
        const retryAfter = response.headers.get('Retry-After') || '60';
        throw new Error(`Rate limited - retry after ${retryAfter} seconds`);
      case 500:
        throw new Error('Internal server error - please retry');
      default:
        throw new Error(`API error ${response.status}: ${errorData.message || 'Unknown error'}`);
    }
  }
}

// Usage
const client = new CompactSaaSClient({
  apiKey: process.env.COMPACTSAAS_API_KEY!,
  baseUrl: process.env.COMPACTSAAS_BASE_URL,
  timeout: 5000,
});

// Track events
await client.track({ event: 'page_view', properties: { page: '/home' } });
await client.track({
  event: 'purchase_completed',
  properties: { amount: 99.99 },
  timestamp: new Date().toISOString(),
});

Python

Installation

# Install requests library
pip install requests

# Optional: For type hints validation
pip install mypy types-requests

Environment Setup

# Set your API key as an environment variable
export COMPACTSAAS_API_KEY="pk_live_your_key_here"

# For staging/development
export COMPACTSAAS_API_KEY="pk_test_your_key_here"
export COMPACTSAAS_BASE_URL="https://api.compactsaasstaging.click/v1"

Basic Client

"""
CompactSaaS API Client for Python
Requires: pip install requests
"""
import os
import requests
from typing import Optional, Dict, Any

# Configuration
API_KEY = os.environ.get('COMPACTSAAS_API_KEY')
BASE_URL = os.environ.get('COMPACTSAAS_BASE_URL', 'https://api.compactsaas.com/v1')

def track_event(
    event: str,
    properties: Optional[Dict[str, Any]] = None,
    timestamp: Optional[str] = None
) -> Dict[str, bool]:
    """
    Track an analytics event.

    Args:
        event: Name of the event (e.g., 'page_view', 'button_click')
        properties: Optional dict of event properties
        timestamp: Optional ISO 8601 timestamp (defaults to server time)

    Returns:
        dict: {'success': True} on success

    Raises:
        requests.HTTPError: If the API returns an error
    """
    payload = {
        'event': event,
        'properties': properties or {},
    }

    if timestamp:
        payload['timestamp'] = timestamp

    response = requests.post(
        f'{BASE_URL}/track',
        headers={
            'x-api-key': API_KEY,
            'Content-Type': 'application/json',
        },
        json=payload,
        timeout=10,
    )

    response.raise_for_status()
    return response.json()

def health_check() -> Dict[str, str]:
    """
    Check API health status.

    Returns:
        dict: {'status': 'ok', 'timestamp': '...', 'version': 'v1'}
    """
    response = requests.get(f'{BASE_URL}/health', timeout=10)
    response.raise_for_status()
    return response.json()

# Usage examples
if __name__ == '__main__':
    # Track a page view
    track_event('page_view', {'page': '/pricing', 'referrer': 'https://google.com'})
    print('Page view tracked')

    # Track a button click
    track_event('button_click', {'button_id': 'signup_cta', 'button_text': 'Get Started'})
    print('Button click tracked')

    # Track a purchase with timestamp
    track_event(
        'purchase_completed',
        {'product_id': 'prod_123', 'amount': 99.99, 'currency': 'USD'},
        timestamp='2024-01-15T10:30:00Z'
    )
    print('Purchase tracked')

    # Check health
    health = health_check()
    print(f"API Status: {health['status']}")

Full Python Client with Error Handling

"""
CompactSaaS API Client - Full-featured Python client with error handling
Requires: pip install requests
"""
import os
import time
import logging
from typing import Optional, Dict, Any, Union
from dataclasses import dataclass
from enum import Enum

import requests
from requests.exceptions import HTTPError, Timeout, RequestException

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('compactsaas')

class ErrorCode(Enum):
    """API error codes"""
    VALIDATION_ERROR = 'VALIDATION_ERROR'
    UNAUTHORIZED = 'UNAUTHORIZED'
    ACCOUNT_DISABLED = 'ACCOUNT_DISABLED'
    RATE_LIMIT_EXCEEDED = 'RATE_LIMIT_EXCEEDED'
    INTERNAL_ERROR = 'INTERNAL_ERROR'
    NOT_FOUND = 'NOT_FOUND'

@dataclass
class CompactSaaSError(Exception):
    """Custom exception for API errors"""
    status_code: int
    error_code: str
    message: str

    def __str__(self):
        return f"CompactSaaSError({self.status_code}): {self.error_code} - {self.message}"

@dataclass
class TrackResult:
    """Result of a track operation"""
    success: bool
    error: Optional[str] = None
    retry_after: Optional[int] = None

class CompactSaaSClient:
    """
    Full-featured CompactSaaS API client with error handling and retries.

    Usage:
        client = CompactSaaSClient(api_key='pk_live_...')
        client.track('page_view', {'page': '/home'})
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: int = 10,
        max_retries: int = 3,
    ):
        """
        Initialize the client.

        Args:
            api_key: API key (defaults to COMPACTSAAS_API_KEY env var)
            base_url: API base URL (defaults to production)
            timeout: Request timeout in seconds
            max_retries: Maximum retry attempts for transient errors
        """
        self.api_key = api_key or os.environ.get('COMPACTSAAS_API_KEY')
        if not self.api_key:
            raise ValueError('API key is required. Set COMPACTSAAS_API_KEY or pass api_key.')

        self.base_url = base_url or os.environ.get(
            'COMPACTSAAS_BASE_URL',
            'https://api.compactsaas.com/v1'
        )
        self.timeout = timeout
        self.max_retries = max_retries

        self._session = requests.Session()
        self._session.headers.update({
            'x-api-key': self.api_key,
            'Content-Type': 'application/json',
        })

    def track(
        self,
        event: str,
        properties: Optional[Dict[str, Any]] = None,
        timestamp: Optional[str] = None,
    ) -> TrackResult:
        """
        Track an analytics event with automatic retries.

        Args:
            event: Event name
            properties: Optional event properties
            timestamp: Optional ISO 8601 timestamp

        Returns:
            TrackResult with success status
        """
        payload = {'event': event, 'properties': properties or {}}
        if timestamp:
            payload['timestamp'] = timestamp

        for attempt in range(self.max_retries):
            try:
                response = self._session.post(
                    f'{self.base_url}/track',
                    json=payload,
                    timeout=self.timeout,
                )

                # Handle rate limiting
                if response.status_code == 429:
                    retry_after = int(response.headers.get('Retry-After', 60))
                    if attempt < self.max_retries - 1:
                        logger.warning(f'Rate limited. Retrying in {retry_after}s...')
                        time.sleep(retry_after)
                        continue
                    return TrackResult(success=False, error='Rate limited', retry_after=retry_after)

                # Handle other errors
                if not response.ok:
                    return self._handle_error(response)

                return TrackResult(success=True)

            except Timeout:
                if attempt < self.max_retries - 1:
                    logger.warning(f'Request timed out. Retrying ({attempt + 1}/{self.max_retries})...')
                    time.sleep(2 ** attempt)  # Exponential backoff
                    continue
                return TrackResult(success=False, error='Request timed out')

            except RequestException as e:
                if attempt < self.max_retries - 1:
                    logger.warning(f'Network error: {e}. Retrying...')
                    time.sleep(2 ** attempt)
                    continue
                return TrackResult(success=False, error=f'Network error: {str(e)}')

        return TrackResult(success=False, error='Max retries exceeded')

    def track_or_raise(
        self,
        event: str,
        properties: Optional[Dict[str, Any]] = None,
        timestamp: Optional[str] = None,
    ) -> Dict[str, bool]:
        """
        Track an event, raising an exception on failure.

        Use this when you need to ensure tracking succeeds.
        """
        payload = {'event': event, 'properties': properties or {}}
        if timestamp:
            payload['timestamp'] = timestamp

        response = self._session.post(
            f'{self.base_url}/track',
            json=payload,
            timeout=self.timeout,
        )

        if not response.ok:
            error_data = response.json() if response.content else {}
            raise CompactSaaSError(
                status_code=response.status_code,
                error_code=error_data.get('code', 'UNKNOWN'),
                message=error_data.get('message', 'Unknown error'),
            )

        return response.json()

    def health(self) -> Dict[str, str]:
        """Check API health status."""
        response = requests.get(f'{self.base_url}/health', timeout=self.timeout)
        response.raise_for_status()
        return response.json()

    def _handle_error(self, response: requests.Response) -> TrackResult:
        """Handle error responses."""
        try:
            error_data = response.json()
        except ValueError:
            error_data = {}

        error_messages = {
            400: f"Validation error: {error_data.get('message', 'Invalid request')}",
            401: 'Invalid or revoked API key',
            403: 'Account disabled - please contact support',
            500: 'Internal server error',
        }

        error_msg = error_messages.get(
            response.status_code,
            f"API error {response.status_code}: {error_data.get('message', 'Unknown error')}"
        )

        return TrackResult(success=False, error=error_msg)

# Usage example
if __name__ == '__main__':
    client = CompactSaaSClient()

    # Safe tracking (returns result, doesn't raise)
    result = client.track('page_view', {'page': '/home'})
    if result.success:
        print('Event tracked successfully')
    else:
        print(f'Failed to track: {result.error}')

    # Strict tracking (raises on failure)
    try:
        client.track_or_raise('purchase_completed', {'amount': 99.99})
        print('Purchase tracked')
    except CompactSaaSError as e:
        print(f'API error: {e}')

cURL

Environment Setup

# Set your API key
export COMPACTSAAS_API_KEY="pk_live_your_key_here"

# Set base URL (optional - defaults to production)
export COMPACTSAAS_BASE_URL="https://api.compactsaas.com/v1"

# For staging/development
export COMPACTSAAS_API_KEY="pk_test_your_key_here"
export COMPACTSAAS_BASE_URL="https://api.compactsaasstaging.click/v1"

Track Event

Basic page view:

curl -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"event": "page_view", "properties": {"page": "/home"}}'

Button click with properties:

curl -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "event": "button_click",
    "properties": {
      "button_id": "signup_cta",
      "button_text": "Get Started",
      "page": "/pricing"
    }
  }'

Purchase with timestamp:

curl -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "event": "purchase_completed",
    "properties": {
      "product_id": "prod_123",
      "product_name": "Pro Plan",
      "amount": 99.99,
      "currency": "USD"
    },
    "timestamp": "2024-01-15T10:30:00Z"
  }'

Health Check

Simple health check (no authentication required):

curl "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/health"

Expected response:

{"status":"ok","timestamp":"2024-01-15T10:30:00.000Z","version":"v1"}

Debugging with Verbose Output

Track event with full request/response details:

curl -v -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"event": "test_event", "properties": {"test": true}}'

Show only HTTP status code:

curl -s -o /dev/null -w "%{http_code}" \
  -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"event": "page_view", "properties": {}}'

Show response with HTTP status:

curl -s -w "\nHTTP Status: %{http_code}\n" \
  -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"event": "page_view", "properties": {"page": "/test"}}'

Testing Error Responses

Test with invalid API key (expect 401):

curl -s -w "\nHTTP Status: %{http_code}\n" \
  -X POST "https://api.compactsaas.com/v1/track" \
  -H "x-api-key: pk_test_invalid_key_12345678901234" \
  -H "Content-Type: application/json" \
  -d '{"event": "test"}'

Test with missing event name (expect 400):

curl -s -w "\nHTTP Status: %{http_code}\n" \
  -X POST "${COMPACTSAAS_BASE_URL:-https://api.compactsaas.com/v1}/track" \
  -H "x-api-key: $COMPACTSAAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"properties": {"page": "/test"}}'

Test with missing API key (expect 403):

curl -s -w "\nHTTP Status: %{http_code}\n" \
  -X POST "https://api.compactsaas.com/v1/track" \
  -H "Content-Type: application/json" \
  -d '{"event": "test"}'

Environment Setup

Recommended Environment Variables

Set these environment variables in your application:

Production vs Staging

Security Best Practices

1. Never hardcode API keys - Always use environment variables
2. Use secrets management - Store keys in AWS Secrets Manager, HashiCorp Vault, etc.
3. Rotate keys regularly - Create a new key, update apps, then revoke the old key
4. Use staging keys for development - Never use production keys in development

Example: Loading from Environment

Node.js:

// Load from environment (recommended)
const apiKey = process.env.COMPACTSAAS_API_KEY;
if (!apiKey) {
  throw new Error('COMPACTSAAS_API_KEY environment variable is required');
}

Python:

import os

api_key = os.environ.get('COMPACTSAAS_API_KEY')
if not api_key:
    raise ValueError('COMPACTSAAS_API_KEY environment variable is required')

Docker:

# Pass API key as environment variable
ENV COMPACTSAAS_API_KEY=""

# Run with API key
docker run -e COMPACTSAAS_API_KEY=pk_live_... myapp

Error Handling Best Practices

Error Response Format

All API errors return a consistent JSON format:

{
  "error": "Error Type",
  "message": "Human-readable description",
  "code": "MACHINE_READABLE_CODE"
}

Error Codes

JavaScript Error Handling

async function trackEventSafe(event, properties) {
  try {
    const response = await fetch(`${BASE_URL}/track`, {
      method: 'POST',
      headers: {
        'x-api-key': API_KEY,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ event, properties }),
    });

    if (!response.ok) {
      const error = await response.json();

      switch (response.status) {
        case 400:
          console.error(`Validation error: ${error.message}`);
          return { success: false, error: 'invalid_request' };
        case 401:
          console.error('Invalid API key - check your configuration');
          return { success: false, error: 'unauthorized' };
        case 403:
          console.error('Account disabled - contact support');
          return { success: false, error: 'account_disabled' };
        case 429:
          const retryAfter = response.headers.get('Retry-After') || 60;
          console.warn(`Rate limited - retry after ${retryAfter}s`);
          return { success: false, error: 'rate_limited', retryAfter };
        default:
          console.error(`API error ${response.status}: ${error.message}`);
          return { success: false, error: 'api_error' };
      }
    }

    return { success: true, data: await response.json() };
  } catch (error) {
    console.error('Network error:', error.message);
    return { success: false, error: 'network_error' };
  }
}

Python Error Handling

from requests.exceptions import HTTPError, Timeout, RequestException

def track_event_safe(event: str, properties: dict = None) -> dict:
    """Track event with comprehensive error handling."""
    try:
        response = requests.post(
            f'{BASE_URL}/track',
            headers={
                'x-api-key': API_KEY,
                'Content-Type': 'application/json',
            },
            json={'event': event, 'properties': properties or {}},
            timeout=10,
        )
        response.raise_for_status()
        return {'success': True, 'data': response.json()}

    except HTTPError as e:
        status = e.response.status_code
        try:
            error_data = e.response.json()
        except ValueError:
            error_data = {}

        error_handlers = {
            400: ('invalid_request', f"Validation error: {error_data.get('message')}"),
            401: ('unauthorized', 'Invalid or revoked API key'),
            403: ('account_disabled', 'Account disabled - contact support'),
            429: ('rate_limited', f"Rate limited - retry after {e.response.headers.get('Retry-After', 60)}s"),
        }

        error_type, message = error_handlers.get(
            status,
            ('api_error', f"API error {status}: {error_data.get('message', 'Unknown')}")
        )

        return {'success': False, 'error': error_type, 'message': message}

    except Timeout:
        return {'success': False, 'error': 'timeout', 'message': 'Request timed out'}

    except RequestException as e:
        return {'success': False, 'error': 'network_error', 'message': str(e)}

Retry Logic

Exponential Backoff

Implement exponential backoff for transient errors (429, 500, network errors):

JavaScript Retry Implementation

/**
 * Track event with exponential backoff retry
 * @param {string} event - Event name
 * @param {Object} properties - Event properties
 * @param {number} maxRetries - Maximum retry attempts (default: 3)
 * @returns {Promise<{success: boolean}>}
 */
async function trackWithRetry(event, properties = {}, maxRetries = 3) {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const response = await fetch(`${BASE_URL}/track`, {
        method: 'POST',
        headers: {
          'x-api-key': API_KEY,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ event, properties }),
      });

      // Handle rate limiting
      if (response.status === 429) {
        const retryAfter = parseInt(response.headers.get('Retry-After') || '60', 10);
        console.log(`Rate limited. Waiting ${retryAfter}s before retry ${attempt + 1}/${maxRetries}`);
        await sleep(retryAfter * 1000);
        continue;
      }

      // Handle server errors with backoff
      if (response.status >= 500) {
        const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s
        console.log(`Server error ${response.status}. Retrying in ${delay}ms...`);
        await sleep(delay);
        continue;
      }

      // Non-retryable errors
      if (!response.ok) {
        const error = await response.json();
        throw new Error(`API error ${response.status}: ${error.message}`);
      }

      return response.json();
    } catch (error) {
      // Network errors - retry with backoff
      if (error.name === 'TypeError' && attempt < maxRetries - 1) {
        const delay = Math.pow(2, attempt) * 1000;
        console.log(`Network error. Retrying in ${delay}ms...`);
        await sleep(delay);
        continue;
      }
      throw error;
    }
  }

  throw new Error(`Failed after ${maxRetries} attempts`);
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

// Usage
try {
  await trackWithRetry('page_view', { page: '/home' });
  console.log('Event tracked successfully');
} catch (error) {
  console.error('Failed to track event:', error.message);
}

Python Retry Implementation

import time
from typing import Optional, Dict, Any

def track_with_retry(
    event: str,
    properties: Optional[Dict[str, Any]] = None,
    max_retries: int = 3,
) -> Dict[str, bool]:
    """
    Track event with exponential backoff retry.

    Args:
        event: Event name
        properties: Event properties
        max_retries: Maximum retry attempts

    Returns:
        API response dict

    Raises:
        Exception: If all retries fail
    """
    last_error = None

    for attempt in range(max_retries):
        try:
            response = requests.post(
                f'{BASE_URL}/track',
                headers={
                    'x-api-key': API_KEY,
                    'Content-Type': 'application/json',
                },
                json={'event': event, 'properties': properties or {}},
                timeout=10,
            )

            # Handle rate limiting
            if response.status_code == 429:
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f'Rate limited. Waiting {retry_after}s before retry {attempt + 1}/{max_retries}')
                time.sleep(retry_after)
                continue

            # Handle server errors with backoff
            if response.status_code >= 500:
                delay = (2 ** attempt)  # 1s, 2s, 4s
                print(f'Server error {response.status_code}. Retrying in {delay}s...')
                time.sleep(delay)
                continue

            # Non-retryable errors
            response.raise_for_status()
            return response.json()

        except requests.exceptions.Timeout:
            delay = (2 ** attempt)
            print(f'Timeout. Retrying in {delay}s...')
            time.sleep(delay)
            last_error = 'Timeout'
            continue

        except requests.exceptions.RequestException as e:
            delay = (2 ** attempt)
            print(f'Network error: {e}. Retrying in {delay}s...')
            time.sleep(delay)
            last_error = str(e)
            continue

    raise Exception(f'Failed after {max_retries} attempts. Last error: {last_error}')

# Usage
try:
    result = track_with_retry('page_view', {'page': '/home'})
    print('Event tracked successfully')
except Exception as e:
    print(f'Failed to track event: {e}')

Batch Tracking

For high-volume applications, consider batching events to reduce API calls:

JavaScript Batch Tracker

/**
 * Batch event tracker - queues events and sends in batches
 */
class BatchTracker {
  constructor(apiKey, options = {}) {
    this.apiKey = apiKey;
    this.baseUrl = options.baseUrl || 'https://api.compactsaas.com/v1';
    this.batchSize = options.batchSize || 10;
    this.flushInterval = options.flushInterval || 5000; // 5 seconds
    this.queue = [];
    this.timer = null;

    // Start flush timer
    this.startTimer();
  }

  /**
   * Queue an event for batch sending
   */
  track(event, properties = {}) {
    this.queue.push({
      event,
      properties,
      timestamp: new Date().toISOString(),
    });

    // Flush if batch size reached
    if (this.queue.length >= this.batchSize) {
      this.flush();
    }
  }

  /**
   * Send all queued events
   */
  async flush() {
    if (this.queue.length === 0) return;

    const events = this.queue.splice(0, this.batchSize);

    // Send events in parallel (respecting rate limits)
    const promises = events.map(event =>
      this.sendEvent(event.event, event.properties, event.timestamp)
    );

    try {
      await Promise.all(promises);
      console.log(`Flushed ${events.length} events`);
    } catch (error) {
      console.error('Batch flush error:', error);
      // Re-queue failed events
      this.queue.unshift(...events);
    }
  }

  /**
   * Send a single event
   */
  async sendEvent(event, properties, timestamp) {
    const response = await fetch(`${this.baseUrl}/track`, {
      method: 'POST',
      headers: {
        'x-api-key': this.apiKey,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ event, properties, timestamp }),
    });

    if (!response.ok) {
      throw new Error(`API error: ${response.status}`);
    }

    return response.json();
  }

  startTimer() {
    this.timer = setInterval(() => this.flush(), this.flushInterval);
  }

  /**
   * Stop the batch tracker and flush remaining events
   */
  async shutdown() {
    if (this.timer) {
      clearInterval(this.timer);
    }
    await this.flush();
  }
}

// Usage
const tracker = new BatchTracker(process.env.COMPACTSAAS_API_KEY, {
  batchSize: 10,
  flushInterval: 5000,
});

// Track events (queued automatically)
tracker.track('page_view', { page: '/home' });
tracker.track('button_click', { button: 'cta' });
tracker.track('page_view', { page: '/pricing' });

// On application shutdown
process.on('SIGTERM', async () => {
  await tracker.shutdown();
  process.exit(0);
});

Python Batch Tracker

import threading
import time
from queue import Queue
from typing import Optional, Dict, Any
import requests

class BatchTracker:
    """
    Batch event tracker - queues events and sends in batches.

    Usage:
        tracker = BatchTracker(api_key='pk_live_...')
        tracker.track('page_view', {'page': '/home'})
        # Events are automatically flushed every 5 seconds or when batch size is reached

        # On shutdown:
        tracker.shutdown()
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = 'https://api.compactsaas.com/v1',
        batch_size: int = 10,
        flush_interval: float = 5.0,
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.queue: Queue = Queue()
        self._shutdown = threading.Event()

        # Start background flush thread
        self._flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
        self._flush_thread.start()

    def track(self, event: str, properties: Optional[Dict[str, Any]] = None):
        """Queue an event for batch sending."""
        from datetime import datetime
        self.queue.put({
            'event': event,
            'properties': properties or {},
            'timestamp': datetime.utcnow().isoformat() + 'Z',
        })

    def flush(self):
        """Send all queued events."""
        events = []
        while not self.queue.empty() and len(events) < self.batch_size:
            try:
                events.append(self.queue.get_nowait())
            except:
                break

        if not events:
            return

        # Send events
        for event_data in events:
            try:
                self._send_event(event_data)
            except Exception as e:
                print(f'Failed to send event: {e}')
                # Re-queue failed event
                self.queue.put(event_data)

        print(f'Flushed {len(events)} events')

    def _send_event(self, event_data: Dict):
        """Send a single event."""
        response = requests.post(
            f'{self.base_url}/track',
            headers={
                'x-api-key': self.api_key,
                'Content-Type': 'application/json',
            },
            json=event_data,
            timeout=10,
        )
        response.raise_for_status()

    def _flush_loop(self):
        """Background thread that flushes events periodically."""
        while not self._shutdown.is_set():
            time.sleep(self.flush_interval)
            if self.queue.qsize() > 0:
                self.flush()

    def shutdown(self):
        """Stop the batch tracker and flush remaining events."""
        self._shutdown.set()
        self.flush()  # Final flush

# Usage
if __name__ == '__main__':
    import os
    import atexit

    tracker = BatchTracker(
        api_key=os.environ['COMPACTSAAS_API_KEY'],
        batch_size=10,
        flush_interval=5.0,
    )

    # Register shutdown handler
    atexit.register(tracker.shutdown)

    # Track events (queued automatically)
    tracker.track('page_view', {'page': '/home'})
    tracker.track('button_click', {'button': 'cta'})
    tracker.track('page_view', {'page': '/pricing'})

    # Keep running for demo
    time.sleep(10)

Quick Reference

Track Event

Health Check

Common Event Names

Related Documentation

• API Reference - Complete endpoint documentation
• Getting Started Guide - Quick introduction
• OpenAPI Specification - Import into Postman, Swagger UI

Support

• Dashboard: compactsaas.com - Manage API keys, view usage
• Email: support@compactsaas.com
• Status Page: status.compactsaas.com