File size: 3,034 Bytes
5e0ae28 |
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 |
"""
Utility decorators for error handling and timing
"""
import time
from functools import wraps
from typing import Callable, Any
import groq
from src.utils.logger import logger
def handle_groq_errors(max_retries: int = 3, retry_delay: float = 1.0) -> Callable:
"""
🛡️ GROQ API ERROR HANDLER WITH EXPONENTIAL BACKOFF
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except groq.RateLimitError as e:
last_exception = e
wait_time = retry_delay * (2 ** attempt)
logger.warning(f"⏳ Rate limit hit. Waiting {wait_time:.1f}s... (Attempt {attempt + 1}/{max_retries})")
time.sleep(wait_time)
except groq.APIConnectionError as e:
last_exception = e
wait_time = retry_delay * (2 ** attempt)
logger.warning(f"🔌 Connection error. Retrying in {wait_time:.1f}s... (Attempt {attempt + 1}/{max_retries})")
time.sleep(wait_time)
except groq.AuthenticationError as e:
logger.error(f"🔑 Authentication failed: {e}")
raise ValueError("Invalid GROQ_API_KEY. Please check your API key.") from e
except groq.BadRequestError as e:
logger.error(f"❌ Invalid request: {e}")
raise ValueError(f"Invalid request parameters: {str(e)}") from e
except Exception as e:
last_exception = e
logger.error(f"❌ Unexpected error: {e}", exc_info=True)
if attempt == max_retries - 1:
break
time.sleep(retry_delay * (2 ** attempt))
error_msg = f"Failed after {max_retries} attempts: {str(last_exception)}"
logger.error(error_msg)
raise Exception(error_msg) from last_exception
return wrapper
return decorator
def with_rate_limit(rate_limiter):
"""
⏱️ RATE LIMITING DECORATOR
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
rate_limiter.acquire()
return func(*args, **kwargs)
return wrapper
return decorator
def timer_decorator(func: Callable) -> Callable:
"""
⏱️ TIMING DECORATOR
"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
result = func(*args, **kwargs)
elapsed_time = time.time() - start_time
logger.debug(f"⏱️ {func.__name__} took {elapsed_time:.2f}s")
return result
return wrapper
|