Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/tenacity/before_sleep.py: 20%

25 statements  

« prev     ^ index     » next       coverage.py v7.4.3, created at 2024-02-26 06:33 +0000

1# Copyright 2016 Julien Danjou 

2# Copyright 2016 Joshua Harlow 

3# Copyright 2013-2014 Ray Holder 

4# 

5# Licensed under the Apache License, Version 2.0 (the "License"); 

6# you may not use this file except in compliance with the License. 

7# You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, software 

12# distributed under the License is distributed on an "AS IS" BASIS, 

13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

14# See the License for the specific language governing permissions and 

15# limitations under the License. 

16 

17import typing 

18 

19from pip._vendor.tenacity import _utils 

20 

21if typing.TYPE_CHECKING: 

22 import logging 

23 

24 from pip._vendor.tenacity import RetryCallState 

25 

26 

27def before_sleep_nothing(retry_state: "RetryCallState") -> None: 

28 """Before call strategy that does nothing.""" 

29 

30 

31def before_sleep_log( 

32 logger: "logging.Logger", 

33 log_level: int, 

34 exc_info: bool = False, 

35) -> typing.Callable[["RetryCallState"], None]: 

36 """Before call strategy that logs to some logger the attempt.""" 

37 

38 def log_it(retry_state: "RetryCallState") -> None: 

39 local_exc_info: BaseException | bool | None 

40 

41 if retry_state.outcome is None: 

42 raise RuntimeError("log_it() called before outcome was set") 

43 

44 if retry_state.next_action is None: 

45 raise RuntimeError("log_it() called before next_action was set") 

46 

47 if retry_state.outcome.failed: 

48 ex = retry_state.outcome.exception() 

49 verb, value = "raised", f"{ex.__class__.__name__}: {ex}" 

50 

51 if exc_info: 

52 local_exc_info = retry_state.outcome.exception() 

53 else: 

54 local_exc_info = False 

55 else: 

56 verb, value = "returned", retry_state.outcome.result() 

57 local_exc_info = False # exc_info does not apply when no exception 

58 

59 if retry_state.fn is None: 

60 # NOTE(sileht): can't really happen, but we must please mypy 

61 fn_name = "<unknown>" 

62 else: 

63 fn_name = _utils.get_callback_name(retry_state.fn) 

64 

65 logger.log( 

66 log_level, 

67 f"Retrying {fn_name} " f"in {retry_state.next_action.sleep} seconds as it {verb} {value}.", 

68 exc_info=local_exc_info, 

69 ) 

70 

71 return log_it