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

1__all__ = ('iscoroutine',) 

2 

3import collections.abc 

4import os 

5import sys 

6import types 

7 

8 

9def _is_debug_mode(): 

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

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

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

13 

14 

15# Prioritize native coroutine check to speed-up 

16# asyncio.iscoroutine. 

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

18_iscoroutine_typecache = set() 

19 

20 

21def iscoroutine(obj): 

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

23 if type(obj) in _iscoroutine_typecache: 

24 return True 

25 

26 if isinstance(obj, _COROUTINE_TYPES): 

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

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

29 # someone stressing the system on purpose. 

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

31 _iscoroutine_typecache.add(type(obj)) 

32 return True 

33 else: 

34 return False 

35 

36 

37def _format_coroutine(coro): 

38 assert iscoroutine(coro) 

39 

40 def get_name(coro): 

41 # Coroutines compiled with Cython sometimes don't have 

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

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

44 # in its __repr__ functions. 

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

46 coro_name = coro.__qualname__ 

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

48 coro_name = coro.__name__ 

49 else: 

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

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

52 return f'{coro_name}()' 

53 

54 def is_running(coro): 

55 try: 

56 return coro.cr_running 

57 except AttributeError: 

58 try: 

59 return coro.gi_running 

60 except AttributeError: 

61 return False 

62 

63 coro_code = None 

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

65 coro_code = coro.cr_code 

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

67 coro_code = coro.gi_code 

68 

69 coro_name = get_name(coro) 

70 

71 if not coro_code: 

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

73 if is_running(coro): 

74 return f'{coro_name} running' 

75 else: 

76 return coro_name 

77 

78 coro_frame = None 

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

80 coro_frame = coro.gi_frame 

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

82 coro_frame = coro.cr_frame 

83 

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

85 # co_filename -- expose that. 

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

87 

88 lineno = 0 

89 

90 if coro_frame is not None: 

91 lineno = coro_frame.f_lineno 

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

93 

94 else: 

95 lineno = coro_code.co_firstlineno 

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

97 

98 return coro_repr