| """ |
| Unified Async API Client - Replace mixed sync/async HTTP calls |
| Implements retry logic, error handling, and logging consistently |
| """ |
|
|
| import aiohttp |
| import asyncio |
| import logging |
| from typing import Optional, Dict, Any, List |
| from datetime import datetime, timedelta |
| import traceback |
|
|
| import config |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class AsyncAPIClient: |
| """ |
| Unified async HTTP client with retry logic and error handling |
| Replaces mixed requests/aiohttp calls throughout the codebase |
| """ |
|
|
| def __init__( |
| self, |
| timeout: int = config.REQUEST_TIMEOUT, |
| max_retries: int = config.MAX_RETRIES, |
| retry_delay: float = 2.0 |
| ): |
| """ |
| Initialize async API client |
| |
| Args: |
| timeout: Request timeout in seconds |
| max_retries: Maximum number of retry attempts |
| retry_delay: Base delay between retries (exponential backoff) |
| """ |
| self.timeout = aiohttp.ClientTimeout(total=timeout) |
| self.max_retries = max_retries |
| self.retry_delay = retry_delay |
| self._session: Optional[aiohttp.ClientSession] = None |
|
|
| async def __aenter__(self): |
| """Async context manager entry""" |
| self._session = aiohttp.ClientSession(timeout=self.timeout) |
| return self |
|
|
| async def __aexit__(self, exc_type, exc_val, exc_tb): |
| """Async context manager exit""" |
| if self._session: |
| await self._session.close() |
|
|
| async def get( |
| self, |
| url: str, |
| params: Optional[Dict[str, Any]] = None, |
| headers: Optional[Dict[str, str]] = None |
| ) -> Optional[Dict[str, Any]]: |
| """ |
| Make async GET request with retry logic |
| |
| Args: |
| url: Request URL |
| params: Query parameters |
| headers: HTTP headers |
| |
| Returns: |
| JSON response as dictionary or None on failure |
| """ |
| if not self._session: |
| raise RuntimeError("Client must be used as async context manager") |
|
|
| for attempt in range(self.max_retries): |
| try: |
| logger.debug(f"GET {url} (attempt {attempt + 1}/{self.max_retries})") |
|
|
| async with self._session.get(url, params=params, headers=headers) as response: |
| response.raise_for_status() |
| data = await response.json() |
| logger.debug(f"GET {url} successful") |
| return data |
|
|
| except aiohttp.ClientResponseError as e: |
| logger.warning(f"HTTP {e.status} error on {url}: {e.message}") |
| if e.status in (404, 400, 401, 403): |
| |
| return None |
| |
| if attempt < self.max_retries - 1: |
| await asyncio.sleep(self.retry_delay * (2 ** attempt)) |
| continue |
| return None |
|
|
| except aiohttp.ClientConnectionError as e: |
| logger.warning(f"Connection error on {url}: {e}") |
| if attempt < self.max_retries - 1: |
| await asyncio.sleep(self.retry_delay * (2 ** attempt)) |
| continue |
| return None |
|
|
| except asyncio.TimeoutError: |
| logger.warning(f"Timeout on {url} (attempt {attempt + 1})") |
| if attempt < self.max_retries - 1: |
| await asyncio.sleep(self.retry_delay * (2 ** attempt)) |
| continue |
| return None |
|
|
| except Exception as e: |
| logger.error(f"Unexpected error on {url}: {e}\n{traceback.format_exc()}") |
| return None |
|
|
| return None |
|
|
| async def post( |
| self, |
| url: str, |
| data: Optional[Dict[str, Any]] = None, |
| json: Optional[Dict[str, Any]] = None, |
| headers: Optional[Dict[str, str]] = None |
| ) -> Optional[Dict[str, Any]]: |
| """ |
| Make async POST request with retry logic |
| |
| Args: |
| url: Request URL |
| data: Form data |
| json: JSON payload |
| headers: HTTP headers |
| |
| Returns: |
| JSON response as dictionary or None on failure |
| """ |
| if not self._session: |
| raise RuntimeError("Client must be used as async context manager") |
|
|
| for attempt in range(self.max_retries): |
| try: |
| logger.debug(f"POST {url} (attempt {attempt + 1}/{self.max_retries})") |
|
|
| async with self._session.post( |
| url, data=data, json=json, headers=headers |
| ) as response: |
| response.raise_for_status() |
| response_data = await response.json() |
| logger.debug(f"POST {url} successful") |
| return response_data |
|
|
| except aiohttp.ClientResponseError as e: |
| logger.warning(f"HTTP {e.status} error on {url}: {e.message}") |
| if e.status in (404, 400, 401, 403): |
| return None |
| if attempt < self.max_retries - 1: |
| await asyncio.sleep(self.retry_delay * (2 ** attempt)) |
| continue |
| return None |
|
|
| except Exception as e: |
| logger.error(f"Error on POST {url}: {e}") |
| if attempt < self.max_retries - 1: |
| await asyncio.sleep(self.retry_delay * (2 ** attempt)) |
| continue |
| return None |
|
|
| return None |
|
|
| async def gather_requests( |
| self, |
| urls: List[str], |
| params_list: Optional[List[Optional[Dict[str, Any]]]] = None |
| ) -> List[Optional[Dict[str, Any]]]: |
| """ |
| Make multiple async GET requests in parallel |
| |
| Args: |
| urls: List of URLs to fetch |
| params_list: Optional list of params for each URL |
| |
| Returns: |
| List of responses (None for failed requests) |
| """ |
| if params_list is None: |
| params_list = [None] * len(urls) |
|
|
| tasks = [ |
| self.get(url, params=params) |
| for url, params in zip(urls, params_list) |
| ] |
|
|
| results = await asyncio.gather(*tasks, return_exceptions=True) |
|
|
| |
| return [ |
| result if not isinstance(result, Exception) else None |
| for result in results |
| ] |
|
|
|
|
| |
|
|
|
|
| async def safe_api_call( |
| url: str, |
| params: Optional[Dict[str, Any]] = None, |
| headers: Optional[Dict[str, str]] = None, |
| timeout: int = config.REQUEST_TIMEOUT |
| ) -> Optional[Dict[str, Any]]: |
| """ |
| Convenience function for single async API call |
| |
| Args: |
| url: Request URL |
| params: Query parameters |
| headers: HTTP headers |
| timeout: Request timeout |
| |
| Returns: |
| JSON response or None on failure |
| """ |
| async with AsyncAPIClient(timeout=timeout) as client: |
| return await client.get(url, params=params, headers=headers) |
|
|
|
|
| async def parallel_api_calls( |
| urls: List[str], |
| params_list: Optional[List[Optional[Dict[str, Any]]]] = None, |
| timeout: int = config.REQUEST_TIMEOUT |
| ) -> List[Optional[Dict[str, Any]]]: |
| """ |
| Convenience function for parallel async API calls |
| |
| Args: |
| urls: List of URLs |
| params_list: Optional params for each URL |
| timeout: Request timeout |
| |
| Returns: |
| List of responses (None for failures) |
| """ |
| async with AsyncAPIClient(timeout=timeout) as client: |
| return await client.gather_requests(urls, params_list) |
|
|
