|
|
""" |
|
|
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) |
|
|
|
