PYTHON
Python: Implementing Robust API Retries with Exponential Backoff
Develop a Python function to make API requests with automatic retries and exponential backoff, effectively managing transient errors and rate limits for stable integrations.
import requests
import time
import logging
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_data_with_retry(url, headers=None, max_retries=5, initial_delay=1, backoff_factor=2):
"""
Fetches data from a given URL with automatic retries and exponential backoff.
Args:
url (str): The API endpoint URL.
headers (dict, optional): Dictionary of HTTP headers. Defaults to None.
max_retries (int): Maximum number of retries for failed requests.
initial_delay (int): Initial delay in seconds before the first retry.
backoff_factor (int): Factor by which the delay increases for subsequent retries.
Returns:
dict: JSON response from the API if successful, None otherwise.
"""
for attempt in range(max_retries + 1):
try:
logging.info(f"Attempt {attempt + 1}/{max_retries + 1}: Fetching data from {url}")
response = requests.get(url, headers=headers, timeout=10) # 10-second timeout
# Check for successful response (200-299)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if 400 <= e.response.status_code < 500 and e.response.status_code != 429:
logging.error(f"Client error ({e.response.status_code}) on {url}: {e.response.text}")
return None # Don't retry client errors (e.g., 401, 404, 403)
elif e.response.status_code == 429:
logging.warning(f"Rate limit hit ({e.response.status_code}) on {url}.")
# Check for 'Retry-After' header if provided by the API
retry_after = e.response.headers.get('Retry-After')
if retry_after:
delay = int(retry_after)
logging.info(f"API suggested to retry after {delay} seconds.")
else:
delay = initial_delay * (backoff_factor ** attempt)
logging.info(f"Waiting for {delay:.2f} seconds before retrying...")
time.sleep(delay)
else: # Server errors (5xx)
logging.error(f"Server error ({e.response.status_code}) on {url}: {e.response.text}")
if attempt < max_retries:
delay = initial_delay * (backoff_factor ** attempt)
logging.info(f"Waiting for {delay:.2f} seconds before retrying...")
time.sleep(delay)
else:
logging.error(f"Max retries reached for {url}.")
return None
except requests.exceptions.ConnectionError as e:
logging.error(f"Connection error on {url}: {e}")
if attempt < max_retries:
delay = initial_delay * (backoff_factor ** attempt)
logging.info(f"Waiting for {delay:.2f} seconds before retrying...")
time.sleep(delay)
else:
logging.error(f"Max retries reached for {url}.")
return None
except requests.exceptions.Timeout:
logging.error(f"Request timed out for {url}.")
if attempt < max_retries:
delay = initial_delay * (backoff_factor ** attempt)
logging.info(f"Waiting for {delay:.2f} seconds before retrying...")
time.sleep(delay)
else:
logging.error(f"Max retries reached for {url}.")
return None
except Exception as e:
logging.error(f"An unexpected error occurred: {e}")
return None
logging.error(f"Failed to fetch data from {url} after {max_retries} retries.")
return None
if __name__ == "__main__":
# Example Usage:
# Replace with a real API endpoint
example_api_url = "https://jsonplaceholder.typicode.com/posts/1"
# To simulate errors, you might use a service like httpstat.us
# example_api_url = "https://httpstat.us/500" # Simulate server error
# example_api_url = "https://httpstat.us/429" # Simulate rate limit
headers = {
'Accept': 'application/json',
# 'Authorization': 'Bearer YOUR_API_TOKEN' # Add if required
}
data = fetch_data_with_retry(example_api_url, headers=headers, max_retries=3, initial_delay=0.5)
if data:
print("
Successfully fetched data:")
print(data)
else:
print("
Failed to fetch data after retries.")
# Another example with a different URL or more retries
# data_2 = fetch_data_with_retry("https://api.example.com/sensitive-data", max_retries=7, initial_delay=2)
# if data_2:
# print("
Successfully fetched sensitive data:")
# print(data_2)
How it works: This Python snippet provides a `fetch_data_with_retry` function that robustly handles API requests, particularly useful for managing transient network errors, server errors (5xx), and rate limits (429). It implements an exponential backoff strategy, increasing the delay between retries. For rate limits, it first checks for a `Retry-After` header provided by the API. Client errors (4xx, excluding 429) are not retried as they usually indicate an issue with the request itself. The function uses the `requests` library for HTTP communication and includes comprehensive error handling and logging to track attempts and failures, making your API integrations more resilient.