Rate Limitinglink

The primary focus of Infraspeak's API is availability and security in support of clients, to ensure fair usage and system stability. So, in order to control the incoming traffic from the API, the Infraspeak API enforces rate limits.

This guide explains how rate limiting works and how to handle it effectively.

Rate Limit Overviewlink

Currently, the limit is 60 requests per minute.

For each API response, we use the following HTTP response headers to provide information about the limit usage:

Exceeding the rate limitlink

Once you reach the rate limit, subsequent requests will get a 429 Too Many Requests HTTP status code response until the 1-minute time window is reset. This means that you need to wait for the period of time to reset in order to execute requests again.

The API returns informative headers when the rate limit exceeds:

Rate Limit Responselink

When you exceed the rate limit, you receive a 429 response:

{
  "status": "error",
  "error": {
    "http_code": 429,
    "message": "Too Many Requests."
  }
}

With headers:

Retry-After: 45
X-RateLimit-Reset: 1779271808

Handling Rate Limitslink

Pythonlink

import requests
import time

class RateLimitHandler:
    def __init__(self, token):
        self.token = token
        self.base_url = "https://api.infraspeak.com/v3"

    def request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint}"
        headers = {"Authorization": f"Bearer {self.token}"}

        while True:
            response = requests.request(
                method, url, headers=headers, **kwargs
            )

            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
                continue

            response.raise_for_status()
            return response.json()

# Usage
client = RateLimitHandler("YOUR_TOKEN")
data = client.request("GET", "failures")

PHPlink

<?php

class RateLimitHandler
{
    private string $token;
    private string $baseUrl = "https://api.infraspeak.com/v3";
    private \GuzzleHttp\Client $client;

    public function __construct(string $token)
    {
        $this->token = $token;
        $this->client = new \GuzzleHttp\Client();
    }

    public function request(string $method, string $endpoint, array $options = []): array
    {
        $url = "{$this->baseUrl}/{$endpoint}";
        $options['headers'] = array_merge(
            $options['headers'] ?? [],
            ['Authorization' => "Bearer {$this->token}"]
        );

        while (true) {
            try {
                $response = $this->client->request($method, $url, $options);

                return json_decode($response->getBody(), true);
            } catch (\GuzzleHttp\Exception\ClientException $e) {
                $response = $e->getResponse();

                if ($response->getStatusCode() === 429) {
                    $retryAfter = (int) ($response->getHeader('Retry-After')[0] ?? 60);

                    echo "Rate limited. Waiting {$retryAfter} seconds...\n";
                    sleep($retryAfter);
                    continue;
                }

                throw $e;
            }
        }
    }
}

// Usage
$client = new RateLimitHandler("YOUR_TOKEN");
$data = $client->request("GET", "failures");

Proactive Rate Limitinglink

Instead of waiting for 429 errors, implement proactive rate limiting:

Token Bucket Algorithmlink

import time
import threading

class RateLimiter:
    def __init__(self, requests_per_minute=60):
        self.rate = requests_per_minute
        self.tokens = requests_per_minute
        self.max_tokens = requests_per_minute
        self.last_update = time.time()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_update
            self.last_update = now

            # Add tokens based on elapsed time
            self.tokens = min(
                self.max_tokens,
                self.tokens + elapsed * (self.rate / 60)
            )

            if self.tokens < 1:
                # Calculate wait time
                wait_time = (1 - self.tokens) / (self.rate / 60)
                time.sleep(wait_time)
                self.tokens = 0
            else:
                self.tokens -= 1

# Usage
limiter = RateLimiter(requests_per_minute=55)  # Leave buffer

def api_request(endpoint):
    limiter.acquire()

    return requests.get(
        f"https://api.infraspeak.com/v3/{endpoint}",
        headers={"Authorization": "Bearer YOUR_TOKEN"}
    )

Simple Delay Patternlink

For batch operations, add delays between requests:

import time

def batch_process(items, api_func, requests_per_minute=55):
    """Process items with rate limiting."""
    delay = 60 / requests_per_minute  # ~1.09 seconds
    results = []

    for i, item in enumerate(items):
        result = api_func(item)
        results.append(result)

        # Progress indicator
        if (i + 1) % 10 == 0:
            print(f"Processed {i + 1}/{len(items)}")

        # Delay before next request (except last)
        if i < len(items) - 1:
            time.sleep(delay)

    return results

# Usage
def create_failure(data):
    return requests.post(
        "https://api.infraspeak.com/v3/failures",
        headers={"Authorization": "Bearer YOUR_TOKEN"},
        json=data
    ).json()

failures_to_create = [{"description": f"Issue {i}"} for i in range(100)]
results = batch_process(failures_to_create, create_failure)

Bulk Operations Strategylink

For large data imports or exports, a strategy of batch with monitoring can be implemented:

import time

class BulkProcessor:
    def __init__(self, client, requests_per_minute=50):
        self.client = client
        self.delay = 60 / requests_per_minute
        self.request_times = []

    def get_current_rate(self):
        """Calculate requests in the last minute."""
        now = time.time()
        self.request_times = [t for t in self.request_times if now - t < 60]
        return len(self.request_times)

    def process_batch(self, items, process_func):
        """Process items with adaptive rate limiting."""
        results = []

        for item in items:
            # Check if we're approaching the limit
            current_rate = self.get_current_rate()

            if current_rate >= 55:
                # Wait until some requests expire
                wait_time = 60 - (time.time() - self.request_times[0]) + 1
                print(f"Approaching limit ({current_rate}/60). Waiting {wait_time:.1f}s...")
                time.sleep(wait_time)

            # Make the request
            self.request_times.append(time.time())
            result = process_func(item)
            results.append(result)

            # Minimum delay between requests
            time.sleep(self.delay)

        return results

Best Practiceslink

1. Leave Buffer Roomlink

Request at 50-55 req/min instead of 60 to account for timing variations:

SAFE_RATE = 55  # Leave 5 requests as buffer
delay = 60 / SAFE_RATE

2. Prioritize Requestslink

During rate limits, prioritize critical requests:

from queue import PriorityQueue

request_queue = PriorityQueue()

# Add requests with priority (lower number = higher priority)
request_queue.put((1, "critical_request"))
request_queue.put((5, "normal_request"))
request_queue.put((10, "background_request"))

3. Use Webhooks for Real-Time Datalink

Instead of polling, subscribe to webhooks:

# Bad - polling every minute
while True:
    check_for_new_failures()  # Uses rate limit
    time.sleep(60)

# Good - receive webhook notifications
# No API calls needed for real-time updates

You can check the Webhooks section for more information.

4. Cache Responseslink

Cache data that doesn't change frequently:

from functools import lru_cache
import time

@lru_cache(maxsize=100)
def get_location_cached(location_id, cache_time):
    """Cache location data for 5 minutes."""
    return api_request("GET", f"locations/{location_id}")

def get_location(location_id):
    # Cache key includes 5-minute bucket
    cache_time = int(time.time() / 300)
    return get_location_cached(location_id, cache_time)