Spaces:
Sleeping
Sleeping
File size: 3,550 Bytes
75788a5 | 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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | 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
|