import logging
import requests
from typing import Dict, Any, Optional
from functools import wraps
from time import sleep
logger = logging.getLogger(__name__)
def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
"""Decorator to retry failed API calls"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
logger.warning(f"Attempt {attempt + 1} failed: {str(e)}. Retrying...")
sleep(delay * (attempt + 1)) # Exponential backoff
return None
return wrapper
return decorator
class BaseService:
"""Base class for all services with common functionality"""
def __init__(self):
self.timeout = 30 # Default timeout in seconds
self._setup_logging()
def _setup_logging(self):
"""Setup logging configuration"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
@retry_on_failure()
def _make_request(
self,
method: str,
url: str,
data: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""Make HTTP request with retry mechanism"""
try:
headers = headers or {"Content-type": "application/json"}
response = requests.request(
method=method,
url=url,
json=data,
headers=headers,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Request failed: {str(e)}")
raise
