python-asyncio:沿着协程链向上走

mat*_*bet 5 python coroutine python-asyncio

我想检查await正在运行的协程中的调用链。我的目标是向我的用户展示一个图表,其中哪些协程在我支持的应用程序中相互调用。但是,一个简单的堆栈是行不通的,因为协程可能会并发运行。

这是我cr_wait用来遍历堆栈的快速测试。我在这里看到了这种方法,但它没有产生我预期的结果。

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()
Run Code Online (Sandbox Code Playgroud)

执行时,此代码仅打印“运行”。我曾(也许是天真地)期望看到:

bar
foo
run
Run Code Online (Sandbox Code Playgroud)

我查看了Task.get_stack,但文档指出此方法将为所有挂起的协程返回单个框架,这似乎没有那么有用。

有没有办法获得完整的协程堆栈?

小智 0

我不知道这是否满足您的问题,但您可以做的是遍历框架:

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()
Run Code Online (Sandbox Code Playgroud)