How do you implement retry logic for failed API requests?
Implementing retry logic for failed API requests is crucial for building resilient applications that can handle temporary network failures, rate limiting, and server errors. This guide covers various retry strategies, implementation patterns, and best practices for handling API failures gracefully.
Understanding Retry Logic Fundamentals
Retry logic is a fault tolerance mechanism that automatically re-attempts failed requests after a specified delay. The key components include:
- Retry conditions: Determining which errors warrant a retry
- Retry strategies: How long to wait between attempts
- Maximum attempts: Preventing infinite retry loops
- Backoff algorithms: Calculating delay intervals
Types of Retry Strategies
1. Fixed Delay Retry
The simplest approach uses a constant delay between retry attempts:
import time
import requests
from requests.exceptions import RequestException
def api_request_with_fixed_retry(url, max_retries=3, delay=1):
"""Make API request with fixed delay retry logic"""
for attempt in range(max_retries + 1):
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except RequestException as e:
if attempt == max_retries:
raise e
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
time.sleep(delay)
return None
2. Exponential Backoff
Exponential backoff increases delay exponentially with each retry attempt, reducing server load:
import time
import random
import requests
from requests.exceptions import RequestException
def api_request_with_exponential_backoff(url, max_retries=3, base_delay=1, max_delay=60):
"""Make API request with exponential backoff retry logic"""
for attempt in range(max_retries + 1):
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except RequestException as e:
if attempt == max_retries:
raise e
# Calculate exponential delay with jitter
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.1) # Add 10% jitter
total_delay = delay + jitter
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {total_delay:.2f}s...")
time.sleep(total_delay)
return None
3. Linear Backoff
Linear backoff increases delay by a constant amount with each retry:
async function apiRequestWithLinearBackoff(url, maxRetries = 3, baseDelay = 1000) {
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
const response = await fetch(url, {
timeout: 10000,
headers: {
'User-Agent': 'MyApp/1.0'
}
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
return await response.json();
} catch (error) {
if (attempt === maxRetries) {
throw error;
}
// Linear backoff: delay increases linearly
const delay = baseDelay * (attempt + 1);
console.log(`Attempt ${attempt + 1} failed: ${error.message}. Retrying in ${delay}ms...`);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
}
Advanced Retry Implementation with Circuit Breaker
For production applications, implement a circuit breaker pattern to prevent cascading failures:
import time
import requests
from enum import Enum
from datetime import datetime, timedelta
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=60, expected_exception=Exception):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time = None
self.state = CircuitState.CLOSED
def call(self, func, *args, **kwargs):
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise e
def _should_attempt_reset(self):
return (datetime.now() - self.last_failure_time) >= timedelta(seconds=self.recovery_timeout)
def _on_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def robust_api_request(url, circuit_breaker, max_retries=3):
"""API request with circuit breaker and retry logic"""
def make_request():
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
for attempt in range(max_retries + 1):
try:
return circuit_breaker.call(make_request)
except Exception as e:
if attempt == max_retries:
raise e
delay = min(2 ** attempt, 30) # Exponential backoff with max 30s
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
time.sleep(delay)
Intelligent Error Classification
Not all errors should trigger retries. Implement smart error classification:
def should_retry_error(exception, status_code=None):
"""Determine if an error should trigger a retry"""
# Network-related errors - always retry
if isinstance(exception, (requests.ConnectionError, requests.Timeout)):
return True
# HTTP status codes that warrant retries
if status_code:
# Server errors (5xx) - retry
if 500 <= status_code < 600:
return True
# Rate limiting (429) - retry with backoff
if status_code == 429:
return True
# Request timeout (408) - retry
if status_code == 408:
return True
# Client errors (4xx) - don't retry (except 429, 408)
if 400 <= status_code < 500:
return False
return False
def smart_retry_request(url, max_retries=3):
"""API request with intelligent error classification"""
for attempt in range(max_retries + 1):
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
if not should_retry_error(e, status_code) or attempt == max_retries:
raise e
except requests.exceptions.RequestException as e:
if not should_retry_error(e) or attempt == max_retries:
raise e
# Calculate delay with exponential backoff
delay = min(2 ** attempt, 60)
print(f"Attempt {attempt + 1} failed. Retrying in {delay}s...")
time.sleep(delay)
Async/Await Implementation in JavaScript
For modern JavaScript applications, implement retry logic with async/await:
class RetryableAPIClient {
constructor(baseURL, options = {}) {
this.baseURL = baseURL;
this.maxRetries = options.maxRetries || 3;
this.baseDelay = options.baseDelay || 1000;
this.maxDelay = options.maxDelay || 30000;
this.retryableStatusCodes = options.retryableStatusCodes || [408, 429, 500, 502, 503, 504];
}
async makeRequest(endpoint, options = {}) {
const url = `${this.baseURL}${endpoint}`;
for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
try {
const response = await fetch(url, {
timeout: 10000,
...options
});
if (response.ok) {
return await response.json();
}
// Check if status code is retryable
if (!this.retryableStatusCodes.includes(response.status)) {
throw new Error(`Non-retryable error: ${response.status} ${response.statusText}`);
}
if (attempt === this.maxRetries) {
throw new Error(`Max retries exceeded: ${response.status} ${response.statusText}`);
}
await this.delay(attempt);
} catch (error) {
if (attempt === this.maxRetries || !this.isRetryableError(error)) {
throw error;
}
console.log(`Attempt ${attempt + 1} failed: ${error.message}`);
await this.delay(attempt);
}
}
}
isRetryableError(error) {
// Network errors, timeouts, and aborted requests are retryable
return error.name === 'TypeError' ||
error.name === 'AbortError' ||
error.message.includes('timeout') ||
error.message.includes('network');
}
async delay(attempt) {
const exponentialDelay = Math.min(this.baseDelay * Math.pow(2, attempt), this.maxDelay);
const jitter = Math.random() * exponentialDelay * 0.1; // 10% jitter
const totalDelay = exponentialDelay + jitter;
console.log(`Waiting ${Math.round(totalDelay)}ms before retry...`);
return new Promise(resolve => setTimeout(resolve, totalDelay));
}
}
// Usage example
const apiClient = new RetryableAPIClient('https://api.example.com', {
maxRetries: 5,
baseDelay: 1000,
maxDelay: 60000
});
try {
const data = await apiClient.makeRequest('/users/123');
console.log('API request successful:', data);
} catch (error) {
console.error('API request failed after all retries:', error);
}
Using Third-Party Libraries
Python: tenacity Library
pip install tenacity
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=60),
retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout))
)
def api_request_with_tenacity(url):
"""API request using tenacity for retry logic"""
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
JavaScript: axios-retry Library
npm install axios axios-retry
const axios = require('axios');
const axiosRetry = require('axios-retry');
// Configure axios with retry logic
axiosRetry(axios, {
retries: 3,
retryDelay: axiosRetry.exponentialDelay,
retryCondition: (error) => {
// Retry on network errors and 5xx status codes
return axiosRetry.isNetworkOrIdempotentRequestError(error) ||
(error.response && error.response.status >= 500);
}
});
async function makeAPIRequest(url) {
try {
const response = await axios.get(url, {
timeout: 10000,
'axios-retry': {
retries: 5,
retryDelay: (retryCount) => {
return Math.min(1000 * Math.pow(2, retryCount), 30000);
}
}
});
return response.data;
} catch (error) {
console.error('API request failed:', error.message);
throw error;
}
}
Best Practices and Considerations
1. Respect Rate Limits
When implementing retry logic, always respect API rate limits and Retry-After headers:
def handle_rate_limit(response):
"""Handle rate limit responses properly"""
if response.status_code == 429:
retry_after = response.headers.get('Retry-After')
if retry_after:
try:
# Retry-After can be in seconds or HTTP date
delay = int(retry_after)
except ValueError:
# Parse HTTP date format
from email.utils import parsedate_to_datetime
retry_time = parsedate_to_datetime(retry_after)
delay = (retry_time - datetime.now()).total_seconds()
print(f"Rate limited. Waiting {delay} seconds...")
time.sleep(max(delay, 0))
return True
return False
2. Implement Logging and Monitoring
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def api_request_with_logging(url, max_retries=3):
"""API request with comprehensive logging"""
for attempt in range(max_retries + 1):
try:
logger.info(f"Making API request to {url} (attempt {attempt + 1})")
response = requests.get(url, timeout=10)
response.raise_for_status()
logger.info(f"API request successful on attempt {attempt + 1}")
return response.json()
except Exception as e:
logger.warning(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == max_retries:
logger.error(f"All retry attempts failed for {url}")
raise e
delay = 2 ** attempt
logger.info(f"Retrying in {delay} seconds...")
time.sleep(delay)
3. Configure Timeouts Appropriately
Set reasonable timeouts to prevent hanging requests:
# Different timeout strategies
requests.get(url, timeout=10) # Single timeout
requests.get(url, timeout=(5, 10)) # Connection timeout: 5s, Read timeout: 10s
When dealing with complex web scraping scenarios that require JavaScript execution, consider integrating retry logic with tools like Puppeteer for handling timeouts in browser automation. Additionally, for applications that need to handle authentication flows, implementing retry logic becomes even more critical to maintain session state across retry attempts.
Testing Retry Logic
Create comprehensive tests for your retry implementations:
import unittest
from unittest.mock import patch, Mock
import requests
class TestRetryLogic(unittest.TestCase):
@patch('requests.get')
def test_successful_retry_after_failure(self, mock_get):
# First call fails, second succeeds
mock_get.side_effect = [
requests.ConnectionError("Connection failed"),
Mock(status_code=200, json=lambda: {"success": True})
]
result = api_request_with_exponential_backoff("http://test.com", max_retries=2)
self.assertEqual(result, {"success": True})
self.assertEqual(mock_get.call_count, 2)
@patch('requests.get')
def test_max_retries_exceeded(self, mock_get):
# All calls fail
mock_get.side_effect = requests.ConnectionError("Connection failed")
with self.assertRaises(requests.ConnectionError):
api_request_with_exponential_backoff("http://test.com", max_retries=2)
self.assertEqual(mock_get.call_count, 3) # Initial + 2 retries
Conclusion
Implementing robust retry logic is essential for building resilient applications that can handle API failures gracefully. The key is to choose appropriate retry strategies based on your specific use case, implement intelligent error classification, and always respect rate limits and server capacity. By combining exponential backoff with jitter, circuit breakers, and proper error handling, you can create applications that maintain high availability even in the face of network instability and temporary service outages.
Remember to monitor your retry logic performance, log retry attempts for debugging, and test various failure scenarios to ensure your implementation works correctly under different conditions.
