Coverage for Lib/asyncio/coroutines.py: 94%

1__all__ = 'iscoroutinefunction', 'iscoroutine' 

2 

3import collections.abc 

4import inspect 

5import os 

6import sys 

7import types 

8 

9 

10def _is_debug_mode(): 

11 # See: https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode. 

12 return sys.flags.dev_mode or (not sys.flags.ignore_environment and 

13 bool(os.environ.get('PYTHONASYNCIODEBUG'))) 

14 

15 

16# A marker for iscoroutinefunction. 

17_is_coroutine = object() 

18 

19 

20def iscoroutinefunction(func): 

21 import warnings 

22 """Return True if func is a decorated coroutine function.""" 

23 warnings._deprecated("asyncio.iscoroutinefunction", 

24 f"{warnings._DEPRECATED_MSG}; " 

25 "use inspect.iscoroutinefunction() instead", 

26 remove=(3,16)) 

27 return _iscoroutinefunction(func) 

28 

29 

30def _iscoroutinefunction(func): 

31 return (inspect.iscoroutinefunction(func) or 

32 getattr(func, '_is_coroutine', None) is _is_coroutine) 

33 

34 

35# Prioritize native coroutine check to speed-up 

36# asyncio.iscoroutine. 

37_COROUTINE_TYPES = (types.CoroutineType, collections.abc.Coroutine) 

38_iscoroutine_typecache = set() 

39 

40 

41def iscoroutine(obj): 

42 """Return True if obj is a coroutine object.""" 

43 if type(obj) in _iscoroutine_typecache: 

44 return True 

45 

46 if isinstance(obj, _COROUTINE_TYPES): 

47 # Just in case we don't want to cache more than 100 

48 # positive types. That shouldn't ever happen, unless 

49 # someone stressing the system on purpose. 

50 if len(_iscoroutine_typecache) < 100: 50 ↛ 52line 50 didn't jump to line 52 because the condition on line 50 was always true

51 _iscoroutine_typecache.add(type(obj)) 

52 return True 

53 else: 

54 return False 

55 

56 

57def _format_coroutine(coro): 

58 assert iscoroutine(coro) 

59 

60 def get_name(coro): 

61 # Coroutines compiled with Cython sometimes don't have 

62 # proper __qualname__ or __name__. While that is a bug 

63 # in Cython, asyncio shouldn't crash with an AttributeError 

64 # in its __repr__ functions. 

65 if hasattr(coro, '__qualname__') and coro.__qualname__: 

66 coro_name = coro.__qualname__ 

67 elif hasattr(coro, '__name__') and coro.__name__: 

68 coro_name = coro.__name__ 

69 else: 

70 # Stop masking Cython bugs, expose them in a friendly way. 

71 coro_name = f'<{type(coro).__name__} without __name__>' 

72 return f'{coro_name}()' 

73 

74 def is_running(coro): 

75 try: 

76 return coro.cr_running 

77 except AttributeError: 

78 try: 

79 return coro.gi_running 

80 except AttributeError: 

81 return False 

82 

83 coro_code = None 

84 if hasattr(coro, 'cr_code') and coro.cr_code: 

85 coro_code = coro.cr_code 

86 elif hasattr(coro, 'gi_code') and coro.gi_code: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 coro_code = coro.gi_code 

88 

89 coro_name = get_name(coro) 

90 

91 if not coro_code: 

92 # Built-in types might not have __qualname__ or __name__. 

93 if is_running(coro): 

94 return f'{coro_name} running' 

95 else: 

96 return coro_name 

97 

98 coro_frame = None 

99 if hasattr(coro, 'gi_frame') and coro.gi_frame: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 coro_frame = coro.gi_frame 

101 elif hasattr(coro, 'cr_frame') and coro.cr_frame: 

102 coro_frame = coro.cr_frame 

103 

104 # If Cython's coroutine has a fake code object without proper 

105 # co_filename -- expose that. 

106 filename = coro_code.co_filename or '<empty co_filename>' 

107 

108 lineno = 0 

109 

110 if coro_frame is not None: 

111 lineno = coro_frame.f_lineno 

112 coro_repr = f'{coro_name} running at {filename}:{lineno}' 

113 

114 else: 

115 lineno = coro_code.co_firstlineno 

116 coro_repr = f'{coro_name} done, defined at {filename}:{lineno}' 

117 

118 return coro_repr