The Buena.ai API uses rate limiting to ensure fair usage and maintain system stability. All limits are applied per API key.

Current Limits

Per Minute

60 requests Short burst protection

Per Hour

1,000 requests Sustained usage limit

Per Day

10,000 requests Daily usage quota

Rate Limit Headers

Every API response includes rate limit information in the headers:

X-RateLimit-Limit: 60
X-RateLimit-Remaining: 59
X-RateLimit-Reset: 1640995200
X-RateLimit-Window: minute

Header Descriptions

HeaderDescription
X-RateLimit-LimitTotal requests allowed in the current window
X-RateLimit-RemainingRequests remaining in the current window
X-RateLimit-ResetUnix timestamp when the current window resets
X-RateLimit-WindowCurrent window type (minute, hour, day)

Rate Limit Exceeded Response

When you exceed a rate limit, the API returns a 429 Too Many Requests response:

{
  "error": true,
  "code": "RATE_LIMIT_EXCEEDED",
  "message": "Rate limit exceeded. Try again in 45 seconds.",
  "version": "2.0",
  "timestamp": "2024-01-20T15:30:00Z",
  "retryAfter": 45,
  "rateLimitInfo": {
    "limit": 60,
    "remaining": 0,
    "reset": 1640995245,
    "window": "minute"
  }
}

Best Practices

Rate Limit Strategies

1. Proactive Monitoring

Monitor your usage with the stats endpoint:

curl -H "x-api-key: YOUR_API_KEY" \
     "https://api.buena.ai/api/v2/keys/stats"

Response:

{
  "success": true,
  "data": {
    "totalKeys": 3,
    "activeKeys": 2,
    "totalUsage": 1250,
    "dailyUsage": {
      "2024-01-20": 45,
      "2024-01-19": 67,
      "2024-01-18": 23
    }
  }
}

2. Request Queuing

Implement a queue system for high-volume operations:

import asyncio
import time
from typing import List, Dict, Any

class BuenaAPIClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.rate_limit = {
            'minute': {'limit': 60, 'remaining': 60, 'reset': 0},
            'hour': {'limit': 1000, 'remaining': 1000, 'reset': 0},
            'day': {'limit': 10000, 'remaining': 10000, 'reset': 0}
        }

    async def make_request(self, endpoint: str, method: str = 'GET', data: Dict = None):
        # Wait if rate limit is low
        await self._wait_if_needed()

        # Make request
        response = await self._http_request(endpoint, method, data)

        # Update rate limit info from headers
        self._update_rate_limits(response.headers)

        return response

    async def _wait_if_needed(self):
        now = time.time()

        for window, limits in self.rate_limit.items():
            if limits['remaining'] <= 1 and limits['reset'] > now:
                wait_time = limits['reset'] - now
                print(f"Rate limit reached for {window}. Waiting {wait_time:.1f}s")
                await asyncio.sleep(wait_time + 1)

3. Intelligent Caching

Cache responses to reduce API calls:

class CachedBuenaClient {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.cache = new Map();
    this.cacheTTL = 5 * 60 * 1000; // 5 minutes
  }

  async getLeads(params = {}) {
    const cacheKey = `leads:${JSON.stringify(params)}`;
    const cached = this.cache.get(cacheKey);

    if (cached && Date.now() - cached.timestamp < this.cacheTTL) {
      return cached.data;
    }

    const response = await this.makeRequest("/leads", { params });
    const data = await response.json();

    this.cache.set(cacheKey, {
      data,
      timestamp: Date.now(),
    });

    return data;
  }
}

Error Handling Examples

JavaScript/Node.js

async function handleRateLimit(apiCall) {
  let retries = 0;
  const maxRetries = 3;

  while (retries < maxRetries) {
    try {
      const response = await apiCall();

      if (response.status === 429) {
        const retryAfter =
          response.headers.get("retry-after") || Math.pow(2, retries);

        console.log(`Rate limited. Retrying in ${retryAfter}s...`);
        await new Promise((resolve) => setTimeout(resolve, retryAfter * 1000));

        retries++;
        continue;
      }

      return response;
    } catch (error) {
      console.error("Request failed:", error);
      retries++;

      if (retries >= maxRetries) throw error;

      await new Promise((resolve) =>
        setTimeout(resolve, Math.pow(2, retries) * 1000)
      );
    }
  }
}

Python

import time
import requests
from typing import Optional

def make_request_with_retry(
    url: str,
    headers: dict,
    max_retries: int = 3,
    base_delay: float = 1.0
) -> Optional[requests.Response]:

    for attempt in range(max_retries):
        try:
            response = requests.get(url, headers=headers)

            if response.status_code == 429:
                retry_after = int(response.headers.get('retry-after', base_delay * (2 ** attempt)))
                print(f"Rate limited. Waiting {retry_after}s before retry {attempt + 1}/{max_retries}")
                time.sleep(retry_after)
                continue

            return response

        except requests.RequestException as e:
            if attempt == max_retries - 1:
                raise e

            delay = base_delay * (2 ** attempt)
            print(f"Request failed. Retrying in {delay}s...")
            time.sleep(delay)

    return None

Monitoring and Alerts

Set up monitoring to track your API usage:

1. Usage Tracking

class UsageTracker {
  constructor() {
    this.usage = {
      minute: 0,
      hour: 0,
      day: 0,
    };
    this.resetTimers();
  }

  trackRequest() {
    this.usage.minute++;
    this.usage.hour++;
    this.usage.day++;

    this.checkThresholds();
  }

  checkThresholds() {
    if (this.usage.minute > 50) {
      console.warn("Approaching minute rate limit");
    }
    if (this.usage.hour > 800) {
      console.warn("Approaching hourly rate limit");
    }
    if (this.usage.day > 8000) {
      console.warn("Approaching daily rate limit");
    }
  }

  resetTimers() {
    setInterval(() => (this.usage.minute = 0), 60 * 1000);
    setInterval(() => (this.usage.hour = 0), 60 * 60 * 1000);
    setInterval(() => (this.usage.day = 0), 24 * 60 * 60 * 1000);
  }
}

2. Alerting

import logging
from datetime import datetime, timedelta

class RateLimitMonitor:
    def __init__(self, alert_threshold=0.8):
        self.alert_threshold = alert_threshold
        self.logger = logging.getLogger(__name__)

    def check_limits(self, headers):
        limit = int(headers.get('X-RateLimit-Limit', 0))
        remaining = int(headers.get('X-RateLimit-Remaining', 0))
        window = headers.get('X-RateLimit-Window', 'unknown')

        usage_percent = (limit - remaining) / limit if limit > 0 else 0

        if usage_percent >= self.alert_threshold:
            self.logger.warning(
                f"High API usage: {usage_percent:.1%} of {window} limit used. "
                f"{remaining} requests remaining."
            )

            # Send alert to monitoring system
            self.send_alert(window, usage_percent, remaining)

    def send_alert(self, window, usage_percent, remaining):
        # Implement your alerting logic here
        # e.g., send to Slack, PagerDuty, etc.
        pass

Need Higher Limits?

If your application requires higher rate limits, contact our support team at support@buena.ai with:

  • Your use case description
  • Expected request volume
  • Current API key prefix
  • Business justification

We offer custom rate limits for enterprise customers and high-volume integrations.