Handling Rate Limits & Quotas

Learn how to gracefully handle rate limits (429) and insufficient credits (402) in your integration.

---

Overview

Circuit enforces two types of limits:

Limit Type	HTTP Status	Description
Rate Limit	429	Too many requests per minute
Credit Quota	402	Insufficient credits to complete operation

Your integration should handle both gracefully to provide a good user experience.

---

Rate Limits

Default Limits

Rate limits are per-API-key and vary by subscription tier:

Tier	Limit (per minute)
Sandbox	25
Starter	100
Growth	500
Enterprise	Custom

Limits are applied per endpoint:

Endpoint	Default Limit
/check-eligibility	Tier limit
/assess-risk	Tier limit
/ingest	3x tier limit (encourages data sharing)
/billing/*	60/min (fixed)

Rate Limit Headers

Every response includes rate limit information:

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 42
X-RateLimit-Reset: 1705329600

Header	Description
X-RateLimit-Limit	Your per-minute limit
X-RateLimit-Remaining	Requests remaining in window
X-RateLimit-Reset	Unix timestamp when window resets

Handling 429 Responses

When rate limited, you'll receive:

HTTP/1.1 429 Too Many Requests
Retry-After: 45
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1705329600

{
  "detail": "Rate limit exceeded",
  "retry_after": 45
}

Always check the Retry-After header - it tells you exactly how long to wait.

---

Implementing Retry Logic

Python Example

import time
from circuit_kyc import CircuitClient, RateLimitError

client = CircuitClient(api_key="sk_sandbox_...")

def check_eligibility_with_retry(email: str, max_retries: int = 3):
    """Check eligibility with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return client.check_eligibility(email=email)

        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise  # Give up after max retries

            # Use Retry-After if provided, otherwise exponential backoff
            wait_time = e.retry_after or (2 ** attempt)
            print(f"Rate limited. Waiting {wait_time}s before retry...")
            time.sleep(wait_time)

# Usage
result = check_eligibility_with_retry("user@example.com")

TypeScript Example

import { CircuitClient, RateLimitError } from '@circuit-kyc/sdk';

const client = new CircuitClient({ apiKey: 'sk_sandbox_...' });

async function checkEligibilityWithRetry(
  email: string,
  maxRetries: number = 3
): Promise<EligibilityResult> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await client.checkEligibility({ email });
    } catch (error) {
      if (error instanceof RateLimitError) {
        if (attempt === maxRetries - 1) throw error;

        const waitTime = error.retryAfter ?? Math.pow(2, attempt);
        console.log(`Rate limited. Waiting ${waitTime}s...`);
        await sleep(waitTime * 1000);
      } else {
        throw error;
      }
    }
  }
  throw new Error('Max retries exceeded');
}

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

---

Credit Quotas (402 Payment Required)

Understanding Credits

Each operation costs credits:

Operation	Credits
check-eligibility	1
assess-risk	5
claim-identity	10
ingest	FREE

When your balance is insufficient, you'll receive:

HTTP/1.1 402 Payment Required

{
  "detail": "Insufficient credits",
  "current_balance": 3,
  "required_credits": 5,
  "operation": "assess-risk"
}

Handling 402 Responses

PythonTypeScript

from circuit_kyc import CircuitClient, InsufficientCreditsError

client = CircuitClient(api_key="sk_sandbox_...")

try:
    risk = client.assess_risk(subject_id="user-123")
except InsufficientCreditsError as e:
    print(f"Need {e.required_credits} credits, have {e.current_balance}")

    # Option 1: Enable auto-recharge
    # Option 2: Redirect to billing page
    # Option 3: Fallback to cheaper operation

    # Fallback example:
    result = client.check_eligibility(email="user@example.com")  # Only 1 credit

import { CircuitClient, InsufficientCreditsError } from '@circuit-kyc/sdk';

const client = new CircuitClient({ apiKey: 'sk_sandbox_...' });

try {
  const risk = await client.assessRisk({ subjectId: 'user-123' });
} catch (error) {
  if (error instanceof InsufficientCreditsError) {
    console.log(`Need ${error.requiredCredits}, have ${error.currentBalance}`);

    // Fallback to cheaper operation
    const result = await client.checkEligibility({ email: 'user@example.com' });
  }
}

---

Proactive Monitoring

Check Balance Before Operations

For expensive operations, check balance first:

def assess_risk_safely(subject_id: str):
    """Check balance before expensive operation."""
    balance = client.get_credit_balance()

    if balance.available_credits < 5:
        # Not enough for assess-risk (5 credits)
        if balance.available_credits >= 1:
            # Fall back to eligibility check
            return client.check_eligibility(subject_id=subject_id)
        else:
            raise Exception("No credits available")

    return client.assess_risk(subject_id=subject_id)

Set Up Low Balance Alerts

Configure webhook alerts when balance drops:

# In your webhook handler
async def process_event(event: dict):
    if event["type"] == "billing.low_balance":
        balance = event["data"]["current_balance"]
        threshold = event["data"]["threshold"]

        # Send alert to your team
        send_slack_alert(f"Low Circuit credits: {balance} (threshold: {threshold})")

        # Or trigger auto-purchase
        # purchase_credits(package="small")

Enable Auto-Recharge

Prevent service interruption by enabling auto-recharge:

# Via SDK (if supported) or API
response = requests.post(
    "https://api.circuit-kyc.com/api/v1/billing/auto-recharge",
    headers={"X-API-Key": api_key},
    json={
        "enabled": True,
        "threshold_credits": 100,  # Recharge when below 100
        "package": "small"  # Purchase 1,000 credits
    }
)

---

Best Practices

1. Implement Circuit Breaker Pattern

Don't hammer the API when it's failing:

import time
from dataclasses import dataclass
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing - reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: int = 60
    failures: int = 0
    state: CircuitState = CircuitState.CLOSED
    last_failure: float = 0

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

    def record_success(self):
        self.failures = 0
        self.state = CircuitState.CLOSED

    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        return True  # HALF_OPEN

2. Use Request Queuing

Queue requests and process at a controlled rate:

import asyncio
from collections import deque

class RequestQueue:
    def __init__(self, rate_limit: int = 100):
        self.queue = deque()
        self.rate_limit = rate_limit
        self.interval = 60 / rate_limit  # seconds between requests

    async def add(self, coro):
        self.queue.append(coro)

    async def process(self):
        while self.queue:
            coro = self.queue.popleft()
            await coro
            await asyncio.sleep(self.interval)

3. Cache Eligibility Results

Don't check the same user repeatedly:

from functools import lru_cache
from datetime import datetime, timedelta

# Simple in-memory cache (use Redis in production)
eligibility_cache = {}
CACHE_TTL = timedelta(hours=1)

def check_eligibility_cached(email: str):
    cache_key = f"eligibility:{email}"

    # Check cache
    if cache_key in eligibility_cache:
        result, timestamp = eligibility_cache[cache_key]
        if datetime.now() - timestamp < CACHE_TTL:
            return result

    # Call API
    result = client.check_eligibility(email=email)

    # Cache result
    eligibility_cache[cache_key] = (result, datetime.now())

    return result

---

Summary

Scenario	Response	Action
Rate limited	429	Wait for Retry-After seconds
Low credits	402	Purchase credits or use cheaper operation
Repeated failures	-	Implement circuit breaker
High volume	-	Queue requests, respect rate limit

Building a robust integration means handling these edge cases gracefully!