import time import json import logging import requests from datetime import datetime from functools import wraps from typing import Dict, Any logger = logging.getLogger(__name__) def with_error_handling(func): """Decorator to add comprehensive error handling.""" @wraps(func) def wrapper(*args, **kwargs): start_time = time.time() try: safe_args = [] for i, arg in enumerate(args[:2]): if isinstance(arg, (str, int, float, bool)): safe_args.append(str(arg)) else: safe_args.append(f"<{type(arg).__name__}>") logger.info(f"Starting {func.__name__} with args: {safe_args}") result = func(*args, **kwargs) if isinstance(result, dict) and 'error' not in result: result = standardize_response(result, func.__name__) execution_time = time.time() - start_time logger.info(f"Completed {func.__name__} in {execution_time:.2f}s") return result except requests.exceptions.Timeout: logger.error(f"Timeout in {func.__name__}") return create_error_response("Request timeout", func.__name__) except requests.exceptions.ConnectionError: logger.error(f"Connection error in {func.__name__}") return create_error_response("Connection failed", func.__name__) except requests.exceptions.RequestException as e: logger.error(f"Request error in {func.__name__}: {e}") return create_error_response(f"Request failed: {str(e)}", func.__name__) except Exception as e: logger.error(f"Unexpected error in {func.__name__}: {e}") return create_error_response(f"Unexpected error: {str(e)}", func.__name__) return wrapper def standardize_response(data: Dict[str, Any], source: str) -> Dict[str, Any]: """Standardize API response format with metadata.""" # Check if data is already a standardized response to avoid double nesting if isinstance(data, dict) and "data" in data and "metadata" in data and "status" in data: return data return { "data": data, "metadata": { "source": source, "timestamp": datetime.now().isoformat(), "version": "1.1.0", "cached": False }, "status": "success" } def create_error_response(error_msg: str, source: str, error_code: str = None) -> Dict[str, Any]: """Create standardized error response with optional error code.""" response = { "data": None, "metadata": { "source": source, "timestamp": datetime.now().isoformat(), "version": "1.1.0" }, "status": "error", "error": error_msg } if error_code: response["error_code"] = error_code return response def make_api_request(url: str, params: Dict[str, Any], timeout: int = 15, max_retries: int = 3) -> requests.Response: """Make API request with retry logic and rate limiting.""" for attempt in range(max_retries): try: time.sleep(0.1 * attempt) response = requests.get(url, params=params, timeout=timeout) if response.status_code == 429: wait_time = 2 ** attempt logger.warning(f"Rate limited, waiting {wait_time}s before retry {attempt + 1}") time.sleep(wait_time) continue return response except requests.exceptions.RequestException as e: if attempt == max_retries - 1: raise logger.warning(f"Request attempt {attempt + 1} failed: {e}") raise requests.exceptions.RequestException("Max retries exceeded") def format_json_output(obj: Any) -> str: """ Formats a Python object as a pretty-printed JSON string. Args: obj: The Python object to format. Returns: A JSON string with an indent of 2 and UTF-8 characters preserved. """ return json.dumps(obj, indent=2, ensure_ascii=False)