Rate Limits

Understanding and working within API rate limits.

Rate Limit Tiers

Tier	Requests/Minute	Requests/Day	Best For
Free	60	10,000	Development, testing
Pro	300	100,000	Production apps
Enterprise	1,000+	Unlimited	High-volume apps

Rate Limit Headers

Every response includes rate limit information:

HTTP/1.1 200 OK
X-RateLimit-Limit: 60
X-RateLimit-Remaining: 45
X-RateLimit-Reset: 1705315860

Header	Description
X-RateLimit-Limit	Max requests per window
X-RateLimit-Remaining	Requests remaining
X-RateLimit-Reset	Unix timestamp when limit resets

When Rate Limited

HTTP/1.1 429 Too Many Requests
Retry-After: 30

{
  "error": "Rate limit exceeded",
  "retry_after": 30
}

Handling Rate Limits

Python

from ofself import OfSelfClient
from ofself.exceptions import RateLimitError
import time

client = OfSelfClient(api_key="your-key")

def fetch_with_rate_limit(func):
    while True:
        try:
            return func()
        except RateLimitError as e:
            wait_time = e.retry_after or 60
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)

JavaScript

import { OfSelfClient, RateLimitError } from '@ofself/sdk';

const client = new OfSelfClient({ apiKey: 'your-key' });

async function fetchWithRateLimit<T>(fn: () => Promise<T>): Promise<T> {
  while (true) {
    try {
      return await fn();
    } catch (error) {
      if (error instanceof RateLimitError) {
        const waitTime = error.retryAfter || 60;
        console.log(`Rate limited. Waiting ${waitTime}s...`);
        await new Promise(r => setTimeout(r, waitTime * 1000));
      } else {
        throw error;
      }
    }
  }
}

Strategies to Avoid Rate Limits

1. Batch Requests

Instead of many small requests, batch when possible:

# ❌ BAD - 100 individual requests
for node_id in node_ids:
    node = client.nodes.get(user_id, node_id)

# ✅ GOOD - One request with filters
nodes = client.nodes.list(user_id, ids=node_ids)

2. Use Pagination Wisely

# ❌ BAD - Fetching one item at a time
for page in range(1, 100):
    result = client.nodes.list(user_id, page=page, per_page=1)

# ✅ GOOD - Fetch in larger batches
result = client.nodes.list(user_id, page=1, per_page=100)

3. Cache Responses

import functools
import time

@functools.lru_cache(maxsize=1000)
def get_node_cached(user_id, node_id):
    return client.nodes.get(user_id, node_id)

# With TTL
from cachetools import TTLCache

cache = TTLCache(maxsize=1000, ttl=300)  # 5 minute TTL

def get_node_with_cache(user_id, node_id):
    key = f"{user_id}:{node_id}"
    if key in cache:
        return cache[key]
    node = client.nodes.get(user_id, node_id)
    cache[key] = node
    return node

4. Use Webhooks Instead of Polling

# ❌ BAD - Polling for changes
while True:
    nodes = client.nodes.list(user_id, updated_after=last_check)
    process_updates(nodes)
    time.sleep(60)

# ✅ GOOD - Use webhooks
# Configure webhook for node.updated events
# Process updates only when they happen

5. Implement Request Queuing

import queue
import threading
import time

class RateLimitedClient:
    def __init__(self, client, requests_per_minute=60):
        self.client = client
        self.interval = 60 / requests_per_minute
        self.last_request = 0
        self.lock = threading.Lock()
    
    def _wait_if_needed(self):
        with self.lock:
            elapsed = time.time() - self.last_request
            if elapsed < self.interval:
                time.sleep(self.interval - elapsed)
            self.last_request = time.time()
    
    def request(self, method, *args, **kwargs):
        self._wait_if_needed()
        return getattr(self.client, method)(*args, **kwargs)

# Usage
rate_limited = RateLimitedClient(client)
result = rate_limited.request('nodes.list', user_id="user-123")

6. Distribute Requests Over Time

import random
import time

def fetch_all_users_data(user_ids):
    results = {}
    for user_id in user_ids:
        # Add jitter to spread requests
        jitter = random.uniform(0, 0.5)
        time.sleep(jitter)
        
        results[user_id] = client.nodes.list(user_id)
    return results

Monitoring Usage

Track your API usage to avoid surprises:

class UsageTracker:
    def __init__(self):
        self.requests_today = 0
        self.reset_time = None
    
    def track(self, response):
        self.requests_today += 1
        remaining = response.headers.get('X-RateLimit-Remaining')
        reset = response.headers.get('X-RateLimit-Reset')
        
        if remaining and int(remaining) < 10:
            print(f"Warning: Only {remaining} requests remaining")
        
        if self.requests_today > 8000:  # 80% of daily limit
            print("Warning: Approaching daily limit")

Rate Limits by Endpoint

Some endpoints have stricter limits:

Endpoint	Limit	Reason
POST /nodes	30/min	Prevents spam
POST /files	10/min	Resource intensive
POST /search/vector	20/min	Compute intensive
GET /nodes	120/min	Higher for reads

Getting Higher Limits

Need more capacity?

1. Upgrade plan - Pro and Enterprise have higher limits
2. Contact us - Explain your use case for custom limits
3. Optimize - Often you can reduce requests with better patterns

Checklist

• Handle 429 responses with retry logic
• Use Retry-After header when available
• Batch requests when possible
• Implement caching
• Use webhooks instead of polling
• Monitor usage and set alerts
• Add jitter to distributed systems