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

1# coding:utf-8 

2 

3import functools 

4import logging 

5import sys 

6import traceback 

7import warnings 

8 

9 

10# Use module-specific logger with a default null handler. 

11_logger = logging.getLogger('backoff') 

12_logger.addHandler(logging.NullHandler()) # pragma: no cover 

13_logger.setLevel(logging.INFO) 

14 

15 

16# Evaluate arg that can be either a fixed value or a callable. 

17def _maybe_call(f, *args, **kwargs): 

18 if callable(f): 

19 try: 

20 return f(*args, **kwargs) 

21 except TypeError: 

22 return f 

23 else: 

24 return f 

25 

26 

27def _init_wait_gen(wait_gen, wait_gen_kwargs): 

28 kwargs = {k: _maybe_call(v) for k, v in wait_gen_kwargs.items()} 

29 initialized = wait_gen(**kwargs) 

30 initialized.send(None) # Initialize with an empty send 

31 return initialized 

32 

33 

34def _next_wait(wait, send_value, jitter, elapsed, max_time): 

35 value = wait.send(send_value) 

36 try: 

37 if jitter is not None: 

38 seconds = jitter(value) 

39 else: 

40 seconds = value 

41 except TypeError: 

42 warnings.warn( 

43 "Nullary jitter function signature is deprecated. Use " 

44 "unary signature accepting a wait value in seconds and " 

45 "returning a jittered version of it.", 

46 DeprecationWarning, 

47 stacklevel=2, 

48 ) 

49 

50 seconds = value + jitter() 

51 

52 # don't sleep longer than remaining allotted max_time 

53 if max_time is not None: 

54 seconds = min(seconds, max_time - elapsed) 

55 

56 return seconds 

57 

58 

59def _prepare_logger(logger): 

60 if isinstance(logger, str): 

61 logger = logging.getLogger(logger) 

62 return logger 

63 

64 

65# Configure handler list with user specified handler and optionally 

66# with a default handler bound to the specified logger. 

67def _config_handlers( 

68 user_handlers, *, default_handler=None, logger=None, log_level=None 

69): 

70 handlers = [] 

71 if logger is not None: 

72 assert log_level is not None, "Log level is not specified" 

73 # bind the specified logger to the default log handler 

74 log_handler = functools.partial( 

75 default_handler, logger=logger, log_level=log_level 

76 ) 

77 handlers.append(log_handler) 

78 

79 if user_handlers is None: 

80 return handlers 

81 

82 # user specified handlers can either be an iterable of handlers 

83 # or a single handler. either way append them to the list. 

84 if hasattr(user_handlers, '__iter__'): 

85 # add all handlers in the iterable 

86 handlers += list(user_handlers) 

87 else: 

88 # append a single handler 

89 handlers.append(user_handlers) 

90 

91 return handlers 

92 

93 

94# Default backoff handler 

95def _log_backoff(details, logger, log_level): 

96 msg = "Backing off %s(...) for %.1fs (%s)" 

97 log_args = [details['target'].__name__, details['wait']] 

98 

99 exc_typ, exc, _ = sys.exc_info() 

100 if exc is not None: 

101 exc_fmt = traceback.format_exception_only(exc_typ, exc)[-1] 

102 log_args.append(exc_fmt.rstrip("\n")) 

103 else: 

104 log_args.append(details['value']) 

105 logger.log(log_level, msg, *log_args) 

106 

107 

108# Default giveup handler 

109def _log_giveup(details, logger, log_level): 

110 msg = "Giving up %s(...) after %d tries (%s)" 

111 log_args = [details['target'].__name__, details['tries']] 

112 

113 exc_typ, exc, _ = sys.exc_info() 

114 if exc is not None: 

115 exc_fmt = traceback.format_exception_only(exc_typ, exc)[-1] 

116 log_args.append(exc_fmt.rstrip("\n")) 

117 else: 

118 log_args.append(details['value']) 

119 

120 logger.log(log_level, msg, *log_args)