Spaces:
Sleeping
Sleeping
File size: 3,876 Bytes
26b9faa | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """
API 重試機制
自動重試失敗的 API 調用，支援指數退避
"""
import asyncio
import functools
import random
from typing import Callable, TypeVar, Any, Optional, Tuple, Type
from core.logging import get_logger
logger = get_logger("core.retry")
T = TypeVar("T")
# 預設重試配置
DEFAULT_MAX_RETRIES = 3
DEFAULT_BASE_DELAY = 1.0 # 秒
DEFAULT_MAX_DELAY = 30.0 # 秒
DEFAULT_EXPONENTIAL_BASE = 2
class RetryConfig:
"""重試配置"""
def __init__(
self,
max_retries: int = DEFAULT_MAX_RETRIES,
base_delay: float = DEFAULT_BASE_DELAY,
max_delay: float = DEFAULT_MAX_DELAY,
exponential_base: float = DEFAULT_EXPONENTIAL_BASE,
jitter: bool = True,
retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
self.retryable_exceptions = retryable_exceptions
def calculate_delay(self, attempt: int) -> float:
"""計算重試延遲（指數退避 + 抖動）"""
delay = self.base_delay * (self.exponential_base ** attempt)
delay = min(delay, self.max_delay)
if self.jitter:
# 添加 ±25% 的隨機抖動
jitter_range = delay * 0.25
delay += random.uniform(-jitter_range, jitter_range)
return max(0, delay)
async def retry_async(
func: Callable[..., Any],
*args,
config: Optional[RetryConfig] = None,
**kwargs,
) -> Any:
"""
異步重試執行函數
Args:
func: 要執行的異步函數
*args: 函數參數
config: 重試配置
**kwargs: 函數關鍵字參數
Returns:
函數執行結果
Raises:
最後一次重試的異常
"""
if config is None:
config = RetryConfig()
last_exception = None
for attempt in range(config.max_retries + 1):
try:
return await func(*args, **kwargs)
except config.retryable_exceptions as e:
last_exception = e
if attempt < config.max_retries:
delay = config.calculate_delay(attempt)
logger.warning(
f"重試 {attempt + 1}/{config.max_retries}: "
f"{func.__name__} 失敗 ({type(e).__name__}: {e})，"
f"{delay:.2f}s 後重試"
)
await asyncio.sleep(delay)
else:
logger.error(
f"重試耗盡: {func.__name__} 在 {config.max_retries} 次重試後仍失敗"
)
raise last_exception
def with_retry(
max_retries: int = DEFAULT_MAX_RETRIES,
base_delay: float = DEFAULT_BASE_DELAY,
retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
"""
重試裝飾器
用法：
@with_retry(max_retries=3)
async def my_api_call():
...
"""
config = RetryConfig(
max_retries=max_retries,
base_delay=base_delay,
retryable_exceptions=retryable_exceptions,
)
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await retry_async(func, *args, config=config, **kwargs)
return wrapper
return decorator
# 預設配置實例
default_retry_config = RetryConfig()
# API 調用專用配置（較短延遲）
api_retry_config = RetryConfig(
max_retries=2,
base_delay=0.5,
max_delay=5.0,
retryable_exceptions=(ConnectionError, TimeoutError, OSError),
)
# 資料庫操作專用配置（較長延遲）
db_retry_config = RetryConfig(
max_retries=3,
base_delay=1.0,
max_delay=10.0,
)
|