import asyncio def _get_stack(coro): ''' Extracts a list of stack frames from a chain of generator/coroutine calls ''' frames = [] while coro: if hasattr(coro, 'cr_frame'): f = coro.cr_frame coro = coro.cr_await elif hasattr(coro, 'ag_frame'): f = coro.ag_frame coro = coro.ag_await elif hasattr(coro, 'gi_frame'): f = coro.gi_frame coro = coro.gi_yieldfrom else: # Note: Can't proceed further. Need the ags_gen or agt_gen attribute # from an asynchronous generator. See https://bugs.python.org/issue32810 f = None coro = None if f is not None: frames.append(f) return frames async def deep(): await asyncio.sleep(1) async def shallow(): await deep() async def surface(): await shallow() # Test 1: A normal chain of coroutines async def main(): await surface() c = main() print(c.send(None)) # Future from asyncio.sleep() print(_get_stack(c)) # Prints the complete stack trace # Test 2: An async generator async def agen(): for i in range(10): await surface() yield i async def amain(): async for n in agen(): print(n) d = amain() print(d.send(None)) # Future from asyncio.sleep() print(_get_stack(d)) # Only top-level frame