X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 847
X-RateLimit-Reset: 1704657600

response = requests.get(
    "https://api.fltr.com/v1/datasets",
    headers={"Authorization": "Bearer YOUR_API_KEY"}
)

print(f"Limit: {response.headers['X-RateLimit-Limit']}")
print(f"Remaining: {response.headers['X-RateLimit-Remaining']}")
print(f"Resets at: {response.headers['X-RateLimit-Reset']}")

{
  "error": "Rate limit exceeded",
  "code": "rate_limit_exceeded",
  "retry_after": 3600
}

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def make_request():
    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        time.sleep(retry_after)
        raise Exception("Rate limited")

    response.raise_for_status()
    return response.json()

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_query(query_hash):
    return query_dataset(query)

# Use hash for cache key
query_hash = hashlib.md5(query.encode()).hexdigest()
results = cached_query(query_hash)

# ❌ Multiple requests
for query in queries:
    results.append(query_dataset(query))  # 10 requests

# ✅ Single batch request
results = batch_query_dataset(queries)  # 1 request

import time

class RateLimitTracker:
    def __init__(self):
        self.requests = []

    def track_request(self, response):
        limit = int(response.headers['X-RateLimit-Limit'])
        remaining = int(response.headers['X-RateLimit-Remaining'])
        reset = int(response.headers['X-RateLimit-Reset'])

        usage_pct = ((limit - remaining) / limit) * 100

        if usage_pct > 80:
            print(f"⚠️ Warning: {usage_pct:.0f}% of rate limit used")

        if remaining == 0:
            wait_time = reset - int(time.time())
            print(f"⛔ Rate limited. Waiting {wait_time}s")
            time.sleep(wait_time)

tracker = RateLimitTracker()

response = requests.get(url, headers=headers)
tracker.track_request(response)

import time

for item in items:
    make_request(item)
    time.sleep(1)  # 1 second delay between requests

# These share the same 1,000/hour limit
api_key_1 = "fltr_sk_abc123..."
api_key_2 = "fltr_sk_def456..."

12:00 - 13:00: 1,000 requests
13:00 - 14:00: 1,000 requests (limit resets)

import logging

def log_rate_limit(response):
    logging.info({
        'rate_limit': response.headers['X-RateLimit-Limit'],
        'remaining': response.headers['X-RateLimit-Remaining'],
        'reset_at': response.headers['X-RateLimit-Reset'],
        'timestamp': time.time()
    })

def check_rate_limit(response):
    remaining = int(response.headers['X-RateLimit-Remaining'])
    limit = int(response.headers['X-RateLimit-Limit'])

    if remaining < limit * 0.1:  # Less than 10% remaining
        send_alert(f"Rate limit warning: {remaining}/{limit}")

def rate_limited_request():
    """Test app behavior when rate limited"""
    class MockResponse:
        status_code = 429
        headers = {'Retry-After': '60'}
        def json(self):
            return {
                'error': 'Rate limit exceeded',
                'retry_after': 60
            }

    return MockResponse()

# Test your error handling
response = rate_limited_request()
handle_rate_limit(response)

import concurrent.futures

def load_test():
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(make_request) for _ in range(100)]

        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                print(f"Error: {e}")

load_test()