Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tenacity/_utils.py: 37%

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. 

16import functools 

17import inspect 

18import sys 

19import typing 

20from datetime import timedelta 

21 

22 

23# sys.maxsize: 

24# An integer giving the maximum value a variable of type Py_ssize_t can take. 

25MAX_WAIT = sys.maxsize / 2 

26 

27 

28class LoggerProtocol(typing.Protocol): 

29 """ 

30 Protocol used by utils expecting a logger (eg: before_log). 

31 

32 Compatible with logging, structlog, loguru, etc... 

33 """ 

34 

35 def log( 

36 self, level: int, msg: str, /, *args: typing.Any, **kwargs: typing.Any 

37 ) -> typing.Any: ... 

38 

39 

40def find_ordinal(pos_num: int) -> str: 

41 # See: https://en.wikipedia.org/wiki/English_numerals#Ordinal_numbers 

42 if pos_num == 0: 

43 return "th" 

44 elif pos_num == 1: 

45 return "st" 

46 elif pos_num == 2: 

47 return "nd" 

48 elif pos_num == 3: 

49 return "rd" 

50 elif 4 <= pos_num <= 20: 

51 return "th" 

52 else: 

53 return find_ordinal(pos_num % 10) 

54 

55 

56def to_ordinal(pos_num: int) -> str: 

57 return f"{pos_num}{find_ordinal(pos_num)}" 

58 

59 

60def get_callback_name(cb: typing.Callable[..., typing.Any]) -> str: 

61 """Get a callback fully-qualified name. 

62 

63 If no name can be produced ``repr(cb)`` is called and returned. 

64 """ 

65 segments = [] 

66 try: 

67 segments.append(cb.__qualname__) 

68 except AttributeError: 

69 try: 

70 segments.append(cb.__name__) 

71 except AttributeError: 

72 pass 

73 if not segments: 

74 return repr(cb) 

75 else: 

76 try: 

77 # When running under sphinx it appears this can be none? 

78 if cb.__module__: 

79 segments.insert(0, cb.__module__) 

80 except AttributeError: 

81 pass 

82 return ".".join(segments) 

83 

84 

85time_unit_type = typing.Union[int, float, timedelta] 

86 

87 

88def to_seconds(time_unit: time_unit_type) -> float: 

89 return float( 

90 time_unit.total_seconds() if isinstance(time_unit, timedelta) else time_unit 

91 ) 

92 

93 

94def is_coroutine_callable(call: typing.Callable[..., typing.Any]) -> bool: 

95 if inspect.isclass(call): 

96 return False 

97 if inspect.iscoroutinefunction(call): 

98 return True 

99 partial_call = isinstance(call, functools.partial) and call.func 

100 dunder_call = partial_call or getattr(call, "__call__", None) 

101 return inspect.iscoroutinefunction(dunder_call) 

102 

103 

104def wrap_to_async_func( 

105 call: typing.Callable[..., typing.Any], 

106) -> typing.Callable[..., typing.Awaitable[typing.Any]]: 

107 if is_coroutine_callable(call): 

108 return call 

109 

110 async def inner(*args: typing.Any, **kwargs: typing.Any) -> typing.Any: 

111 return call(*args, **kwargs) 

112 

113 return inner