Understanding API Rate Limiting and Working Around It with Python#

1
import requests
2
import time
3

4
api_url = "https://api.example.com/data"
5
api_key = "YOUR_API_KEY" # Replace with your actual API key
6

7
headers = {
8
    "Authorization": f"Bearer {api_key}", # Common way to pass API key
9
    # Add other necessary headers
10
}
11

12
try:
13
    response = requests.get(api_url, headers=headers)
14

15
    # Check status code
16
    if response.status_code == 200:
17
        print("Request successful.")
18
        data = response.json()
19
        # Process data...
20

21
        # Check for rate limit headers (if provided by the API)
22
        limit = response.headers.get('X-RateLimit-Limit')
23
        remaining = response.headers.get('X-RateLimit-Remaining')
24
        reset_time = response.headers.get('X-RateLimit-Reset') # May be timestamp or seconds
25

26
        print(f"Rate Limit Info:")
27
        print(f"  Limit: {limit}")
28
        print(f"  Remaining: {remaining}")
29
        print(f"  Reset: {reset_time}") # Need to interpret based on API docs
30

31
    elif response.status_code == 429:
32
        print("Rate limit exceeded.")
33
        # Handle rate limit - look for Retry-After header
34
        retry_after = response.headers.get('Retry-After')
35
        if retry_after:
36
            wait_time = int(retry_after)
37
            print(f"Waiting for {wait_time} seconds as per Retry-After header.")
38
            time.sleep(wait_time)
39
            # After waiting, you would typically retry the request
40
        else:
41
            print("No Retry-After header found. Waiting for a default period.")
42
            time.sleep(60) # Default wait, adjust based on API knowledge
43

44
    else:
45
        print(f"Request failed with status code: {response.status_code}")
46
        print(f"Response body: {response.text}")
47

48
except requests.exceptions.RequestException as e:
49
    print(f"An error occurred during the request: {e}")

1
import requests
2
import time
3

4
api_url_template = "https://api.example.com/data/{item_id}"
5
api_key = "YOUR_API_KEY"
6
item_ids = range(1000) # Example: need to fetch data for 1000 items
7

8
headers = {
9
    "Authorization": f"Bearer {api_key}",
10
}
11

12
requests_per_minute = 60 # Example API limit
13
delay_between_requests = 60.0 / requests_per_minute # Calculate minimum delay
14

15
print(f"Calculated delay: {delay_between_requests:.2f} seconds between requests.")
16

17
for item_id in item_ids:
18
    url = api_url_template.format(item_id=item_id)
19
    try:
20
        response = requests.get(url, headers=headers)
21

22
        if response.status_code == 200:
23
            print(f"Successfully fetched item {item_id}")
24
            # Process data...
25
        elif response.status_code == 429:
26
            print(f"Rate limit hit fetching item {item_id}. Handling...")
27
            # Simple throttling alone might not prevent this, but let's add
28
            # a wait based on Retry-After if available, otherwise a longer pause.
29
            retry_after = response.headers.get('Retry-After')
30
            wait_time = int(retry_after) if retry_after else 60 # Default wait 60s
31
            print(f"Waiting {wait_time} seconds.")
32
            time.sleep(wait_time)
33
            # Note: A more robust solution would retry the *failed* request here.
34
        else:
35
            print(f"Failed to fetch item {item_id} with status {response.status_code}")
36
            # Decide how to handle other errors (log, skip, etc.)
37

38
    except requests.exceptions.RequestException as e:
39
        print(f"An error occurred fetching item {item_id}: {e}")
40
        # Decide how to handle network errors (log, retry, etc.)
41

42
    # Implement the calculated delay *after* processing the request (or failure)
43
    print(f"Waiting {delay_between_requests:.2f} seconds before next request.")
44
    time.sleep(delay_between_requests)
45

46
print("Finished processing items.")

1
import requests
2
import time
3
import random
4

5
def make_throttled_request(url, headers, max_retries=5, initial_delay=1.0):
6
    """
7
    Makes an API request with retries and exponential backoff on rate limits (429)
8
    or common request exceptions.
9
    """
10
    retry_count = 0
11
    while retry_count < max_retries:
12
        try:
13
            response = requests.get(url, headers=headers)
14

15
            if response.status_code == 200:
16
                # Success
17
                return response
18

19
            elif response.status_code == 429:
20
                retry_count += 1
21
                print(f"Rate limit hit ({response.status_code}). Retry attempt {retry_count}/{max_retries} for {url}")
22

23
                retry_after = response.headers.get('Retry-After')
24
                if retry_after:
25
                    # API specifies wait time
26
                    wait_time = int(retry_after)
27
                    print(f"  Waiting {wait_time} seconds as per Retry-After header.")
28
                    time.sleep(wait_time)
29
                else:
30
                    # Use exponential backoff with jitter
31
                    # Base delay increases exponentially: initial_delay * (2 ^ (retry_count - 1))
32
                    # Add jitter: multiply by a random factor between 0.5 and 1.5 (example jitter range)
33
                    calculated_delay = initial_delay * (2 ** (retry_count - 1))
34
                    jitter = random.uniform(0.5, 1.5)
35
                    wait_time = calculated_delay * jitter
36
                    print(f"  No Retry-After. Waiting with backoff+jitter: {wait_time:.2f} seconds.")
37
                    time.sleep(wait_time)
38

39
            else:
40
                # Handle other non-success status codes
41
                print(f"Request failed with status code {response.status_code} for {url}")
42
                # Decide whether to retry on other codes or return the response
43
                # For simplicity, let's return non-429 errors immediately
44
                return response
45

46
        except requests.exceptions.RequestException as e:
47
            retry_count += 1
48
            print(f"Request exception: {e}. Retry attempt {retry_count}/{max_retries} for {url}")
49
            # Implement backoff for network errors as well
50
            calculated_delay = initial_delay * (2 ** (retry_count - 1))
51
            jitter = random.uniform(0.5, 1.5)
52
            wait_time = calculated_delay * jitter
53
            print(f"  Waiting with backoff+jitter: {wait_time:.2f} seconds.")
54
            time.sleep(wait_time)
55

56
    # If loop finishes, max retries reached without success
57
    print(f"Max retries ({max_retries}) reached for {url}. Failing request.")
58
    return None # Or raise an exception
59

60
# --- Example Usage ---
61
api_url_template = "https://api.example.com/data/{item_id}"
62
api_key = "YOUR_API_KEY"
63
item_ids = range(100) # Example: fetching 100 items
64

65
headers = {
66
    "Authorization": f"Bearer {api_key}",
67
}
68

69
results = {}
70
for item_id in item_ids:
71
    url = api_url_template.format(item_id=item_id)
72
    response = make_throttled_request(url, headers)
73

74
    if response and response.status_code == 200:
75
        results[item_id] = response.json()
76
        print(f"Processed item {item_id}")
77
    elif response is not None: # Handle explicit non-200 errors returned
78
         print(f"Skipping item {item_id} due to explicit error status: {response.status_code}")
79
    else: # Handle cases where max retries were reached
80
        print(f"Skipping item {item_id} after multiple retries.")
81

82
print("\nFinished processing items.")
83
# print("Results:", results) # Uncomment to see fetched data

1
import requests
2
import time
3
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception, retry_if_result
4

5
api_url_template = "https://api.example.com/data/{item_id}"
6
api_key = "YOUR_API_KEY"
7

8
headers = {
9
    "Authorization": f"Bearer {api_key}",
10
}
11

12
# Helper function to check if the response indicates rate limit
13
def is_rate_limited(response):
14
    return response.status_code == 429
15

16
# Helper function to check if the response is not successful (anything but 200)
17
def is_not_successful(response):
18
    return response.status_code != 200
19

20
@retry(
21
    wait=wait_exponential(multiplier=1, min=1, max=60), # Wait 1s, 2s, 4s, 8s... up to 60s max
22
    stop=stop_after_attempt(5), # Retry up to 5 times
23
    retry=retry_if_result(is_rate_limited) | retry_if_exception(requests.exceptions.RequestException)
24
)
25
def fetch_item_with_retry(item_id):
26
    """Fetches a single item, retrying on rate limit or network issues."""
27
    url = api_url_template.format(item_id=item_id)
28
    print(f"Attempting to fetch item {item_id}...")
29
    response = requests.get(url, headers=headers)
30

31
    if response.status_code == 429:
32
        print(f"  Rate limit hit for item {item_id}. Retrying...")
33
        # tenacity will handle the wait based on the decorator configuration
34
        # If Retry-After header is present, tenacity can also be configured to use it
35
        # (requires custom wait strategy or helper function)
36
        # For simplicity here, we rely on the exponential backoff defined in @retry
37
        return response # Return response so retry_if_result can check status
38

39
    elif response.status_code != 200:
40
        print(f"  Request failed with status {response.status_code} for item {item_id}. Not retrying this status.")
41
        # Returning the response will prevent retry if retry_if_result is used with is_not_successful
42
        # Raising an exception would trigger retry_if_exception
43
        return response
44

45
    print(f"  Successfully fetched item {item_id}.")
46
    return response # Success, retry stops
47

48
# --- Example Usage with tenacity ---
49
item_ids = range(100) # Example: fetching 100 items
50
results = {}
51

52
for item_id in item_ids:
53
    try:
54
        response = fetch_item_with_retry(item_id)
55
        if response and response.status_code == 200:
56
            results[item_id] = response.json()
57
        elif response:
58
             print(f"Skipping item {item_id} due to final status: {response.status_code}")
59
        else: # Should not happen with stop_after_attempt, but good practice
60
             print(f"Skipping item {item_id} after retries failed.")
61

62
    except Exception as e: # Catch exceptions raised by tenacity after max retries
63
        print(f"Failed to fetch item {item_id} after multiple retries: {e}")
64
        # Handle permanent failure for this item
65

66
print("\nFinished processing items.")
67
# print("Results:", results) # Uncomment to see fetched data