import logging from functools import wraps from typing import Callable, Any import time logger = logging.getLogger(__name__) class RateLimiter: """Simple rate limiter for API calls.""" def __init__(self, max_calls: int = 10, time_window: int = 60): self.max_calls = max_calls self.time_window = time_window self.calls = [] def is_allowed(self, key: str) -> bool: """Check if call is allowed within rate limit.""" current_time = time.time() self.calls = [ (k, t) for k, t in self.calls if current_time - t < self.time_window ] key_calls = [t for k, t in self.calls if k == key] if len(key_calls) >= self.max_calls: return False self.calls.append((key, current_time)) return True def get_wait_time(self, key: str) -> float: """Get wait time in seconds before next allowed call.""" current_time = time.time() key_calls = [ t for k, t in self.calls if k == key and current_time - t < self.time_window ] if len(key_calls) < self.max_calls: return 0.0 oldest_call = min(key_calls) wait_time = self.time_window - (current_time - oldest_call) return max(0.0, wait_time) rate_limiter = RateLimiter(max_calls=20, time_window=60) def with_retry(max_retries: int = 3, delay: float = 1.0): """Decorator to retry function on failure.""" 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 Exception as e: last_exception = e logger.warning( f"Attempt {attempt + 1}/{max_retries} failed for {func.__name__}: {e}" ) if attempt < max_retries - 1: time.sleep(delay * (attempt + 1)) logger.error(f"All {max_retries} attempts failed for {func.__name__}") raise last_exception return wrapper return decorator def with_fallback(fallback_value: Any = None): """Decorator to return fallback value on error.""" def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: try: return func(*args, **kwargs) except Exception as e: logger.error(f"Error in {func.__name__}: {e}. Returning fallback.") return fallback_value return wrapper return decorator def log_execution_time(func: Callable) -> Callable: """Decorator to log function execution time.""" @wraps(func) def wrapper(*args, **kwargs) -> Any: start_time = time.time() result = func(*args, **kwargs) end_time = time.time() execution_time = end_time - start_time logger.info(f"{func.__name__} executed in {execution_time:.2f}s") return result return wrapper class APIError(Exception): """Custom API error.""" pass class RateLimitError(Exception): """Rate limit exceeded error.""" pass class CacheError(Exception): """Cache operation error.""" pass