import time
import requests
from typing import Optional, Dict, Any
def fetch_with_retry(
url: str,
headers: Dict[str, str],
max_retries: int = 3
) -> Dict[str, Any]:
"""Make API request with exponential backoff retry logic."""
retryable_errors = ['bad_gateway', 'service_unavailable', 'internal_server_error', 'rate_limit_exceeded']
for attempt in range(max_retries + 1):
try:
response = requests.get(url, headers=headers)
if response.ok:
return response.json()
error = response.json()
error_code = error.get('code', '')
# Check if error is retryable
if error_code in retryable_errors and attempt < max_retries:
if error_code == 'rate_limit_exceeded':
# Use rate limit header if available
delay = int(response.headers.get('X-RateLimit-Reset', 60))
else:
# Exponential backoff: 1s, 2s, 4s
delay = (2 ** attempt)
print(f"Retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
time.sleep(delay)
continue
# Non-retryable error
raise Exception(f"API Error: {error_code} - {error.get('message', 'Unknown error')}")
except requests.exceptions.RequestException as e:
# Network errors are retryable
if attempt < max_retries:
delay = 2 ** attempt
print(f"Network error, retrying in {delay}s")
time.sleep(delay)
continue
raise
# Usage
try:
customer = fetch_with_retry(
'https://api.platform.dakota.xyz/customers/123',
headers={'x-api-key': api_key}
)
print('Customer:', customer)
except Exception as e:
print(f'Request failed after retries: {e}')