Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/backoff/_async.py: 11%

1# coding:utf-8 

2import datetime 

3import functools 

4import asyncio 

5from datetime import timedelta 

6 

7from backoff._common import (_init_wait_gen, _maybe_call, _next_wait) 

8 

9 

10def _ensure_coroutine(coro_or_func): 

11 if asyncio.iscoroutinefunction(coro_or_func): 

12 return coro_or_func 

13 else: 

14 @functools.wraps(coro_or_func) 

15 async def f(*args, **kwargs): 

16 return coro_or_func(*args, **kwargs) 

17 return f 

18 

19 

20def _ensure_coroutines(coros_or_funcs): 

21 return [_ensure_coroutine(f) for f in coros_or_funcs] 

22 

23 

24async def _call_handlers(handlers, 

25 *, 

26 target, args, kwargs, tries, elapsed, 

27 **extra): 

28 details = { 

29 'target': target, 

30 'args': args, 

31 'kwargs': kwargs, 

32 'tries': tries, 

33 'elapsed': elapsed, 

34 } 

35 details.update(extra) 

36 for handler in handlers: 

37 await handler(details) 

38 

39 

40def retry_predicate(target, wait_gen, predicate, 

41 *, 

42 max_tries, max_time, jitter, 

43 on_success, on_backoff, on_giveup, 

44 wait_gen_kwargs): 

45 on_success = _ensure_coroutines(on_success) 

46 on_backoff = _ensure_coroutines(on_backoff) 

47 on_giveup = _ensure_coroutines(on_giveup) 

48 

49 # Easy to implement, please report if you need this. 

50 assert not asyncio.iscoroutinefunction(max_tries) 

51 assert not asyncio.iscoroutinefunction(jitter) 

52 

53 assert asyncio.iscoroutinefunction(target) 

54 

55 @functools.wraps(target) 

56 async def retry(*args, **kwargs): 

57 

58 # update variables from outer function args 

59 max_tries_value = _maybe_call(max_tries) 

60 max_time_value = _maybe_call(max_time) 

61 

62 tries = 0 

63 start = datetime.datetime.now() 

64 wait = _init_wait_gen(wait_gen, wait_gen_kwargs) 

65 while True: 

66 tries += 1 

67 elapsed = timedelta.total_seconds(datetime.datetime.now() - start) 

68 details = { 

69 "target": target, 

70 "args": args, 

71 "kwargs": kwargs, 

72 "tries": tries, 

73 "elapsed": elapsed, 

74 } 

75 

76 ret = await target(*args, **kwargs) 

77 if predicate(ret): 

78 max_tries_exceeded = (tries == max_tries_value) 

79 max_time_exceeded = (max_time_value is not None and 

80 elapsed >= max_time_value) 

81 

82 if max_tries_exceeded or max_time_exceeded: 

83 await _call_handlers(on_giveup, **details, value=ret) 

84 break 

85 

86 try: 

87 seconds = _next_wait(wait, ret, jitter, elapsed, 

88 max_time_value) 

89 except StopIteration: 

90 await _call_handlers(on_giveup, **details, value=ret) 

91 break 

92 

93 await _call_handlers(on_backoff, **details, value=ret, 

94 wait=seconds) 

95 

96 # Note: there is no convenient way to pass explicit event 

97 # loop to decorator, so here we assume that either default 

98 # thread event loop is set and correct (it mostly is 

99 # by default), or Python >= 3.5.3 or Python >= 3.6 is used 

100 # where loop.get_event_loop() in coroutine guaranteed to 

101 # return correct value. 

102 # See for details: 

103 # <https://groups.google.com/forum/#!topic/python-tulip/yF9C-rFpiKk> 

104 # <https://bugs.python.org/issue28613> 

105 await asyncio.sleep(seconds) 

106 continue 

107 else: 

108 await _call_handlers(on_success, **details, value=ret) 

109 break 

110 

111 return ret 

112 

113 return retry 

114 

115 

116def retry_exception(target, wait_gen, exception, 

117 *, 

118 max_tries, max_time, jitter, giveup, 

119 on_success, on_backoff, on_giveup, raise_on_giveup, 

120 wait_gen_kwargs): 

121 on_success = _ensure_coroutines(on_success) 

122 on_backoff = _ensure_coroutines(on_backoff) 

123 on_giveup = _ensure_coroutines(on_giveup) 

124 giveup = _ensure_coroutine(giveup) 

125 

126 # Easy to implement, please report if you need this. 

127 assert not asyncio.iscoroutinefunction(max_tries) 

128 assert not asyncio.iscoroutinefunction(jitter) 

129 

130 @functools.wraps(target) 

131 async def retry(*args, **kwargs): 

132 

133 max_tries_value = _maybe_call(max_tries) 

134 max_time_value = _maybe_call(max_time) 

135 

136 tries = 0 

137 start = datetime.datetime.now() 

138 wait = _init_wait_gen(wait_gen, wait_gen_kwargs) 

139 while True: 

140 tries += 1 

141 elapsed = timedelta.total_seconds(datetime.datetime.now() - start) 

142 details = { 

143 "target": target, 

144 "args": args, 

145 "kwargs": kwargs, 

146 "tries": tries, 

147 "elapsed": elapsed, 

148 } 

149 

150 try: 

151 ret = await target(*args, **kwargs) 

152 except exception as e: 

153 giveup_result = await giveup(e) 

154 max_tries_exceeded = (tries == max_tries_value) 

155 max_time_exceeded = (max_time_value is not None and 

156 elapsed >= max_time_value) 

157 

158 if giveup_result or max_tries_exceeded or max_time_exceeded: 

159 await _call_handlers(on_giveup, **details, exception=e) 

160 if raise_on_giveup: 

161 raise 

162 return None 

163 

164 try: 

165 seconds = _next_wait(wait, e, jitter, elapsed, 

166 max_time_value) 

167 except StopIteration: 

168 await _call_handlers(on_giveup, **details, exception=e) 

169 raise e 

170 

171 await _call_handlers(on_backoff, **details, wait=seconds, 

172 exception=e) 

173 

174 # Note: there is no convenient way to pass explicit event 

175 # loop to decorator, so here we assume that either default 

176 # thread event loop is set and correct (it mostly is 

177 # by default), or Python >= 3.5.3 or Python >= 3.6 is used 

178 # where loop.get_event_loop() in coroutine guaranteed to 

179 # return correct value. 

180 # See for details: 

181 # <https://groups.google.com/forum/#!topic/python-tulip/yF9C-rFpiKk> 

182 # <https://bugs.python.org/issue28613> 

183 await asyncio.sleep(seconds) 

184 else: 

185 await _call_handlers(on_success, **details) 

186 

187 return ret 

188 return retry