Coverage for qreward / utils / retry.py: 97.14%

1import asyncio (empty)

2import functools (empty)

3import random (empty)

4import time (empty)

5from collections.abc import Callable (empty)

6from typing import ( (empty)

7 Any, 

8 Union, 

9 Iterable, 

10) 

11 

12from qreward.types import RetryPredicate (empty)

13 

14 

15def retry( (empty)

16 *, 

17 max_retries: int = 3, 

18 delay: float = 1.0, 

19 backoff_factor: float = 2.0, 

20 jitter: bool = True, 

21 retry_on: Union[ 

22 type[Exception], 

23 Iterable[type[Exception]], 

24 RetryPredicate, 

25 ] = Exception, 

26 check_exception: RetryPredicate | None = None, 

27): 

28 """重试装饰器, 支持同步和异步函数。 

29 

30 参数: 

31 max_retries: 最大重试次数(不包括首次调用) 

32 delay: 初始延迟时间(单位: 秒) 

33 backoff_factor: 指数退避因子 

34 jitter: 是否添加随机抖动(0~1倍delay的随机值) 

35 retry_on: 指定要重试的异常类型或异常判断函数 

36 check_exception: 自定义异常判断函数, 接收异常实例, 返回是否重试 

37 

38 示例: 

39 @retry( 

40 max_retries=3, 

41 delay=0.1, 

42 retry_on=(ValueError, ConnectionError), 

43 ) 

44 async def fetch_data(): 

45 ... 

46 

47 @retry(check_exception=lambda e: isinstance(e, ValueError) 

48 and "retry" in str(e)) 

49 def unreliable_func(): 

50 ... 

51 """ 

52 

53 def should_retry(exception: Exception) -> bool: (empty)

54 # 检查 retry_on 

55 if callable(retry_on): (empty)

56 if not retry_on(exception): (empty)

57 return False (empty)

58 elif not isinstance( (empty)

59 exception, 

60 tuple(retry_on) if isinstance(retry_on, (list, tuple)) 

61 else (retry_on,), 

62 ): 

63 return False (empty)

64 

65 # 检查 check_exception 

66 if check_exception is not None: (empty)

67 return check_exception(exception) (empty)

68 

69 return True (empty)

70 

71 def exponential_backoff(attempt: int) -> float: (empty)

72 # 计算指数退避时间 

73 exp_delay = delay * (backoff_factor ** (attempt - 1)) (empty)

74 if jitter: (empty)

75 exp_delay += random.uniform(0, delay) (empty)

76 return exp_delay (empty)

77 

78 def decorator(func: Callable[..., Any]) -> Callable[..., Any]: (empty)

79 is_coroutine = asyncio.iscoroutinefunction(func) (empty)

80 

81 @functools.wraps(func) (empty)

82 async def async_wrapper(*args, **kwargs): (empty)

83 last_exc = None (empty)

84 for attempt in range(1, max_retries + 2): 84 ↛ 93line 84 didn't jump to line 93 because the loop on line 84 didn't complete(empty)

85 try: (empty)

86 return await func(*args, **kwargs) (empty)

87 except Exception as e: (empty)

88 last_exc = e (empty)

89 if attempt >= max_retries + 1 or not should_retry(e): (empty)

90 break (empty)

91 wait_time = exponential_backoff(attempt) (empty)

92 await asyncio.sleep(wait_time) (empty)

93 raise last_exc (empty)

94 

95 @functools.wraps(func) (empty)

96 def sync_wrapper(*args, **kwargs): (empty)

97 last_exc = None (empty)

98 for attempt in range(1, max_retries + 2): 98 ↛ 107line 98 didn't jump to line 107 because the loop on line 98 didn't complete(empty)

99 try: (empty)

100 return func(*args, **kwargs) (empty)

101 except Exception as e: (empty)

102 last_exc = e (empty)

103 if attempt >= max_retries + 1 or not should_retry(e): (empty)

104 break (empty)

105 wait_time = exponential_backoff(attempt) (empty)

106 time.sleep(wait_time) (empty)

107 raise last_exc (empty)

108 

109 # 根据函数类型返回对应的包装器 

110 return async_wrapper if is_coroutine else sync_wrapper (empty)

111 

112 return decorator (empty)