PYTHON
Resilient API Calls: Implementing a Retry Mechanism in Python
Build a robust retry mechanism for Python API calls using exponential backoff, handling transient errors to improve application resilience and reliability in integrations.
import requests
import time
from requests.exceptions import RequestException, HTTPError
def make_api_call_with_retry(
url,
method="GET",
headers=None,
json_payload=None,
data_payload=None,
params=None,
max_retries=3,
backoff_factor=0.5,
status_forcelist=(500, 502, 503, 504),
timeout=10
):
"""
Makes an API call with a retry mechanism for transient errors.
Args:
url (str): The API endpoint URL.
method (str): HTTP method (e.g., 'GET', 'POST').
headers (dict, optional): Request headers.
json_payload (dict, optional): JSON data to send in the request body.
data_payload (dict, optional): Form data to send in the request body.
params (dict, optional): Query parameters.
max_retries (int): Maximum number of retries.
backoff_factor (float): Factor for exponential backoff (delay = backoff_factor * (2 ** (retry_attempt - 1))).
status_forcelist (tuple): HTTP status codes to retry on.
timeout (int): Request timeout in seconds.
Returns:
requests.Response: The successful response object.
Raises:
requests.exceptions.RequestException: If all retry attempts fail.
"""
for retry_attempt in range(max_retries + 1):
try:
response = requests.request(
method,
url,
headers=headers,
json=json_payload,
data=data_payload,
params=params,
timeout=timeout
)
response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
return response
except (RequestException, HTTPError) as e:
is_retryable = isinstance(e, HTTPError) and e.response.status_code in status_forcelist or \
isinstance(e, RequestException) and not isinstance(e, HTTPError) # Network errors are generally retryable
if retry_attempt < max_retries and is_retryable:
wait_time = backoff_factor * (2 ** retry_attempt)
print(f"API call failed ({e}). Retrying in {wait_time:.2f} seconds... (Attempt {retry_attempt + 1}/{max_retries})")
time.sleep(wait_time)
else:
print(f"API call failed after {retry_attempt} attempts: {e}")
raise # Re-raise the last exception if retries exhausted or not retryable error
# Example Usage:
# try:
# # Example for a GET request
# # response = make_api_call_with_retry(
# # "https://api.example.com/sometimes-fails",
# # method="GET",
# # max_retries=5,
# # backoff_factor=1,
# # status_forcelist=(500, 502, 503, 504, 429) # Add 429 for rate limiting
# # )
# # print("Successful GET response:", response.json())
# # Example for a POST request
# # response = make_api_call_with_retry(
# # "https://api.example.com/create",
# # method="POST",
# # json_payload={"name": "test", "value": 123},
# # max_retries=3
# # )
# # print("Successful POST response:", response.json())
# except RequestException as e:
# print(f"Final failure after all retries: {e}")
How it works: This Python function provides a robust wrapper for `requests` to implement a retry mechanism with exponential backoff. It automatically retries API calls that encounter transient network errors or specific HTTP status codes (e.g., 5xx server errors, 429 rate limits). The `backoff_factor` ensures that retry attempts are spaced out, preventing overwhelming the API server during recovery and enhancing the overall reliability of your integrations.