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

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 

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

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

30 if pos_num == 0: 

31 return "th" 

32 elif pos_num == 1: 

33 return "st" 

34 elif pos_num == 2: 

35 return "nd" 

36 elif pos_num == 3: 

37 return "rd" 

38 elif 4 <= pos_num <= 20: 

39 return "th" 

40 else: 

41 return find_ordinal(pos_num % 10) 

42 

43 

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

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

46 

47 

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

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

50 

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

52 """ 

53 segments = [] 

54 try: 

55 segments.append(cb.__qualname__) 

56 except AttributeError: 

57 try: 

58 segments.append(cb.__name__) 

59 except AttributeError: 

60 pass 

61 if not segments: 

62 return repr(cb) 

63 else: 

64 try: 

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

66 if cb.__module__: 

67 segments.insert(0, cb.__module__) 

68 except AttributeError: 

69 pass 

70 return ".".join(segments) 

71 

72 

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

74 

75 

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

77 return float( 

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

79 ) 

80 

81 

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

83 if inspect.isclass(call): 

84 return False 

85 if inspect.iscoroutinefunction(call): 

86 return True 

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

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

89 return inspect.iscoroutinefunction(dunder_call)