1__all__ = 'iscoroutinefunction', 'iscoroutine' 2 3import collections.abc 4import inspect 5import os 6import sys 7import traceback 8import types 9 10 11def _is_debug_mode(): 12 # See: https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode. 13 return sys.flags.dev_mode or (not sys.flags.ignore_environment and 14 bool(os.environ.get('PYTHONASYNCIODEBUG'))) 15 16 17# A marker for iscoroutinefunction. 18_is_coroutine = object() 19 20 21def iscoroutinefunction(func): 22 """Return True if func is a decorated coroutine function.""" 23 return (inspect.iscoroutinefunction(func) or 24 getattr(func, '_is_coroutine', None) is _is_coroutine) 25 26 27# Prioritize native coroutine check to speed-up 28# asyncio.iscoroutine. 29_COROUTINE_TYPES = (types.CoroutineType, types.GeneratorType, 30 collections.abc.Coroutine) 31_iscoroutine_typecache = set() 32 33 34def iscoroutine(obj): 35 """Return True if obj is a coroutine object.""" 36 if type(obj) in _iscoroutine_typecache: 37 return True 38 39 if isinstance(obj, _COROUTINE_TYPES): 40 # Just in case we don't want to cache more than 100 41 # positive types. That shouldn't ever happen, unless 42 # someone stressing the system on purpose. 43 if len(_iscoroutine_typecache) < 100: 44 _iscoroutine_typecache.add(type(obj)) 45 return True 46 else: 47 return False 48 49 50def _format_coroutine(coro): 51 assert iscoroutine(coro) 52 53 def get_name(coro): 54 # Coroutines compiled with Cython sometimes don't have 55 # proper __qualname__ or __name__. While that is a bug 56 # in Cython, asyncio shouldn't crash with an AttributeError 57 # in its __repr__ functions. 58 if hasattr(coro, '__qualname__') and coro.__qualname__: 59 coro_name = coro.__qualname__ 60 elif hasattr(coro, '__name__') and coro.__name__: 61 coro_name = coro.__name__ 62 else: 63 # Stop masking Cython bugs, expose them in a friendly way. 64 coro_name = f'<{type(coro).__name__} without __name__>' 65 return f'{coro_name}()' 66 67 def is_running(coro): 68 try: 69 return coro.cr_running 70 except AttributeError: 71 try: 72 return coro.gi_running 73 except AttributeError: 74 return False 75 76 coro_code = None 77 if hasattr(coro, 'cr_code') and coro.cr_code: 78 coro_code = coro.cr_code 79 elif hasattr(coro, 'gi_code') and coro.gi_code: 80 coro_code = coro.gi_code 81 82 coro_name = get_name(coro) 83 84 if not coro_code: 85 # Built-in types might not have __qualname__ or __name__. 86 if is_running(coro): 87 return f'{coro_name} running' 88 else: 89 return coro_name 90 91 coro_frame = None 92 if hasattr(coro, 'gi_frame') and coro.gi_frame: 93 coro_frame = coro.gi_frame 94 elif hasattr(coro, 'cr_frame') and coro.cr_frame: 95 coro_frame = coro.cr_frame 96 97 # If Cython's coroutine has a fake code object without proper 98 # co_filename -- expose that. 99 filename = coro_code.co_filename or '<empty co_filename>' 100 101 lineno = 0 102 103 if coro_frame is not None: 104 lineno = coro_frame.f_lineno 105 coro_repr = f'{coro_name} running at {filename}:{lineno}' 106 107 else: 108 lineno = coro_code.co_firstlineno 109 coro_repr = f'{coro_name} done, defined at {filename}:{lineno}' 110 111 return coro_repr 112