""" HTTP Client wrapper with retry logic and error handling for API integrations. """ import requests import time import json from typing import Dict, Any, Optional, Union from urllib.parse import urljoin import logging logger = logging.getLogger(__name__) class HTTPClientError(Exception): """Base exception for HTTP client errors.""" pass class HTTPTimeoutError(HTTPClientError): """Exception raised when HTTP request times out.""" pass class HTTPRetryExhaustedError(HTTPClientError): """Exception raised when all retry attempts are exhausted.""" pass class HTTPClient: """ HTTP client wrapper with retry logic, error handling, and authentication support. Supports all HTTP methods (GET, POST, PUT, DELETE) with configurable retry policies. """ def __init__(self, base_url: Optional[str] = None, timeout: int = 30, default_headers: Optional[Dict[str, str]] = None): """ Initialize HTTP client. Args: base_url: Base URL for all requests (optional) timeout: Request timeout in seconds default_headers: Default headers to include in all requests """ self.base_url = base_url self.timeout = timeout self.default_headers = default_headers or {} self.session = requests.Session() # Set default headers on session self.session.headers.update(self.default_headers) def _build_url(self, url: str) -> str: """Build full URL from base URL and endpoint.""" if self.base_url and not url.startswith(('http://', 'https://')): return urljoin(self.base_url, url) return url def _prepare_request_kwargs(self, headers: Optional[Dict[str, str]] = None, auth: Optional[tuple] = None, **kwargs) -> Dict[str, Any]: """Prepare request keyword arguments.""" request_kwargs = { 'timeout': self.timeout, **kwargs } # Merge headers if headers: merged_headers = self.default_headers.copy() merged_headers.update(headers) request_kwargs['headers'] = merged_headers # Add authentication if auth: request_kwargs['auth'] = auth return request_kwargs def _execute_request_with_retry(self, method: str, url: str, retry_attempts: int = 3, retry_delay: float = 1.0, backoff_factor: float = 2.0, **kwargs) -> requests.Response: """ Execute HTTP request with retry logic. Args: method: HTTP method (GET, POST, PUT, DELETE) url: Request URL retry_attempts: Number of retry attempts retry_delay: Initial delay between retries in seconds backoff_factor: Multiplier for delay between retries **kwargs: Additional request arguments Returns: requests.Response object Raises: HTTPTimeoutError: If request times out HTTPRetryExhaustedError: If all retry attempts are exhausted HTTPClientError: For other HTTP errors """ last_exception = None current_delay = retry_delay for attempt in range(retry_attempts + 1): try: logger.debug(f"HTTP {method} request to {url}, attempt {attempt + 1}") response = self.session.request(method, url, **kwargs) # Check for HTTP errors if response.status_code >= 400: if response.status_code >= 500 and attempt < retry_attempts: logger.warning(f"{response.json()}") # Retry on server errors logger.warning(f"Server error {response.status_code}, retrying in {current_delay}s") time.sleep(current_delay) current_delay *= backoff_factor continue else: # Don't retry on client errors (4xx) or if no retries left response.raise_for_status() logger.debug(f"HTTP {method} request successful, status: {response.status_code}") return response except requests.exceptions.Timeout as e: last_exception = HTTPTimeoutError(f"Request timed out after {self.timeout}s") logger.warning(f"Request timeout, attempt {attempt + 1}/{retry_attempts + 1}") except requests.exceptions.ConnectionError as e: last_exception = HTTPClientError(f"Connection error: {str(e)}") logger.warning(f"Connection error, attempt {attempt + 1}/{retry_attempts + 1}") except requests.exceptions.HTTPError as e: last_exception = HTTPClientError(f"HTTP error: {str(e)}") logger.error(f"HTTP error: {e}") break # Don't retry on HTTP errors except Exception as e: last_exception = HTTPClientError(f"Unexpected error: {str(e)}") logger.error(f"Unexpected error: {e}") break # Sleep before retry (except on last attempt) if attempt < retry_attempts: logger.debug(f"Retrying in {current_delay}s...") time.sleep(current_delay) current_delay *= backoff_factor # All retries exhausted raise HTTPRetryExhaustedError(f"All {retry_attempts + 1} attempts failed. Last error: {last_exception}") def get(self, url: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, auth: Optional[tuple] = None, retry_attempts: int = 3, retry_delay: float = 1.0, **kwargs) -> requests.Response: """ Execute GET request. Args: url: Request URL params: Query parameters headers: Request headers auth: Authentication tuple (username, password) retry_attempts: Number of retry attempts retry_delay: Initial delay between retries **kwargs: Additional request arguments Returns: requests.Response object """ full_url = self._build_url(url) request_kwargs = self._prepare_request_kwargs( headers=headers, auth=auth, params=params, **kwargs ) return self._execute_request_with_retry( 'GET', full_url, retry_attempts, retry_delay, **request_kwargs ) def post(self, url: str, data: Optional[Union[Dict[str, Any], str]] = None, json_data: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, auth: Optional[tuple] = None, retry_attempts: int = 3, retry_delay: float = 1.0, **kwargs) -> requests.Response: """ Execute POST request. Args: url: Request URL data: Request body data json_data: JSON data to send (will set Content-Type header) headers: Request headers auth: Authentication tuple (username, password) retry_attempts: Number of retry attempts retry_delay: Initial delay between retries **kwargs: Additional request arguments Returns: requests.Response object """ full_url = self._build_url(url) request_kwargs = self._prepare_request_kwargs( headers=headers, auth=auth, **kwargs ) if json_data is not None: request_kwargs['json'] = json_data elif data is not None: request_kwargs['data'] = data return self._execute_request_with_retry( 'POST', full_url, retry_attempts, retry_delay, **request_kwargs ) def put(self, url: str, data: Optional[Union[Dict[str, Any], str]] = None, json_data: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, auth: Optional[tuple] = None, retry_attempts: int = 3, retry_delay: float = 1.0, **kwargs) -> requests.Response: """ Execute PUT request. Args: url: Request URL data: Request body data json_data: JSON data to send (will set Content-Type header) headers: Request headers auth: Authentication tuple (username, password) retry_attempts: Number of retry attempts retry_delay: Initial delay between retries **kwargs: Additional request arguments Returns: requests.Response object """ full_url = self._build_url(url) request_kwargs = self._prepare_request_kwargs( headers=headers, auth=auth, **kwargs ) if json_data is not None: request_kwargs['json'] = json_data elif data is not None: request_kwargs['data'] = data return self._execute_request_with_retry( 'PUT', full_url, retry_attempts, retry_delay, **request_kwargs ) def delete(self, url: str, headers: Optional[Dict[str, str]] = None, auth: Optional[tuple] = None, retry_attempts: int = 3, retry_delay: float = 1.0, **kwargs) -> requests.Response: """ Execute DELETE request. Args: url: Request URL headers: Request headers auth: Authentication tuple (username, password) retry_attempts: Number of retry attempts retry_delay: Initial delay between retries **kwargs: Additional request arguments Returns: requests.Response object """ full_url = self._build_url(url) request_kwargs = self._prepare_request_kwargs( headers=headers, auth=auth, **kwargs ) return self._execute_request_with_retry( 'DELETE', full_url, retry_attempts, retry_delay, **request_kwargs ) def close(self): """Close the HTTP session.""" self.session.close() def __enter__(self): """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit.""" self.close()