python-asyncio: walk up the coroutine chain

user_3238702 0
Viewed 476

I'd like to inspect the chain of await calls from within a running coroutine. My goal is to present a graph to my users of which coroutines called each other from within the application I support. But, a simple stack won't work because coroutines may run concurrently.

Here's a quick test I did using cr_wait to walk the stack. I saw this approach here, but it didn't produce the results I expected.

from asyncio import get_event_loop
from asyncio import Task


def main():
    loop = get_event_loop()
    loop.run_until_complete(run())


async def run():
    await foo()


async def foo():
    await bar()


async def bar():
    print_coro_stack(Task.current_task()._coro)


def print_coro_stack(coro):
    if coro is None:
        return
    print(coro.__name__)
    if coro.cr_await:
        print_coro_stack(coro.cr_await)


if __name__ == '__main__':
    main()

When executed, this code prints only "run". I had (perhaps naively) expected to see:

bar
foo
run

I looked into Task.get_stack, but the documentation notes that this method will return a single frame for all suspended coroutines, which doesn't seem that useful.

Is there a way I can get the complete coroutine stack?

python-asyncio coroutine python
1 Answers

I don't know if this satisfies your question but what you could do is traverse the frames:

from asyncio import get_event_loop
from asyncio import Task
import inspect

def main():

    loop = get_event_loop()
    loop.run_until_complete(run())


async def run():
    await foo()


async def foo():
    await bar()

async def bar():
    print_frame_stack(inspect.currentframe())


def print_frame_stack(frame):
    if frame is None:
        return
    print(frame.f_code.co_name)
    if frame.f_back:
        print_frame_stack(frame.f_back)


if __name__ == '__main__':
    main()
user_96250500
Related