Spaces:
Running
Running
File size: 3,843 Bytes
a04c9e9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
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."""
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) -> Dict[str, Any]:
"""Create standardized error response."""
return {
"data": None,
"metadata": {
"source": source,
"timestamp": datetime.now().isoformat(),
"version": "1.1.0"
},
"status": "error",
"error": error_msg
}
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) |