Python中如何等待动态生成的异步代码执行完毕？

你当前用await asyncio.sleep(0)的方式无法保证生成的异步函数完全执行完毕，因为sleep(0)仅触发一次事件循环切换，只能处理当前就绪的任务步骤，无法覆盖异步函数内所有后续的异步调用。以下是可靠的解决方法：

方法1：暴露任务对象并直接await #

修改生成代码的逻辑，让创建的任务对象能被调用代码访问，然后直接await该任务：

生成的代码示例 #

import asyncio

async def my_app():
    # 示例异步逻辑
    await asyncio.sleep(0.5)
    print("第一步完成")
    await asyncio.sleep(0.5)
    print("第二步完成")

loop = asyncio.get_event_loop()
task = loop.create_task(my_app())

调用代码示例 #

# 用自定义命名空间执行exec，避免污染全局变量
exec_namespace = {}
exec(my_code, exec_namespace)
# 直接await生成的任务，确保其所有异步步骤执行完毕
await exec_namespace['task']

这种方式能精准等待my_app内所有异步调用完成，是最可靠的方案。

方法2：使用run_until_complete（兼容旧版Python） #

如果你的Python版本低于3.7，且调用代码处于同步上下文（非异步函数内），可以用事件循环的run_until_complete方法等待任务结束：

exec_namespace = {}
exec(my_code, exec_namespace)
loop = exec_namespace['loop']
loop.run_until_complete(exec_namespace['task'])

但注意：如果调用者本身已经在异步函数中，不要使用这个方法，会导致事件循环嵌套报错。

原方法不可靠的原因 #

await asyncio.sleep(0)仅让事件循环处理当前已就绪的任务片段，比如my_app中第一个await之前的代码会执行，但第一个await触发后任务会进入等待状态，此时sleep(0)执行完毕，调用者会继续往下走，无法等待后续的异步步骤完成。

内容的提问来源于stack exchange，提问作者W1nTer003