You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多
文档控制台
免费开始使用

如何在可同时作为Jupyter Notebook运行的Python脚本中使用asyncio?

解决Jupyter与脚本模式下的异步代码兼容问题

我之前也碰到过完全一样的麻烦——用带# %%分隔符的.py文件兼顾Jupyter交互式运行和普通脚本模式时,异步代码总是两边“水土不服”。试过nest-asyncio踩了兼容性坑,调autoawait也没解决问题,后来摸索出两个靠谱的方案,分享给你:

方案1:主动检测运行环境

通过判断当前是否处于Jupyter环境,分支执行对应的异步调用逻辑:

# %%
import asyncio

def is_running_in_jupyter():
    try:
        from IPython import get_ipython
        # 检查是否存在IPython实例,且是Jupyter内核环境
        return get_ipython() is not None and 'IPKernelApp' in get_ipython().config
    except ImportError:
        # 没装IPython肯定不是Jupyter环境
        return False

async def my_async_func():
    await asyncio.sleep(1)
    return "搞定异步兼容!"

# %%
if is_running_in_jupyter():
    # Jupyter环境下直接用顶层await(依赖默认开启的autoawait)
    result = await my_async_func()
else:
    # 普通脚本环境用asyncio.run启动事件循环
    result = asyncio.run(my_async_func())

print(result)

这个方案逻辑清晰,通过IPython的配置明确判断环境,不需要依赖异常捕获来做分支处理。

方案2:被动捕获异常兼容

如果不想引入IPython的依赖,可以直接通过asyncio.run()的异常来判断当前是否已有运行中的事件循环:

# %%
import asyncio

async def my_async_func():
    await asyncio.sleep(1)
    return "搞定异步兼容!"

# %%
try:
    # 先尝试用脚本模式的方式运行
    result = asyncio.run(my_async_func())
except RuntimeError as e:
    # 如果报错是"无法从运行中的事件循环调用",说明是Jupyter环境
    if "cannot be called from a running event loop" in str(e):
        result = await my_async_func()
    else:
        # 其他异常正常抛出,不吞错误
        raise

print(result)

这个方案更通用,哪怕是在IPython终端这类支持顶层await的环境也能正常工作,不需要额外导入库。

为什么之前的方案行不通?

  • nest-asyncio:确实容易和一些依赖原生事件循环的库冲突,比如某些异步数据库驱动或爬虫框架,我之前用的时候也踩过坑;
  • 关闭autoawait:Jupyter的顶层await依赖这个特性,关掉之后反而需要手动管理事件循环,反而更麻烦;
  • shebang声明:只能强制用IPython运行脚本,但没法解决普通Python解释器对顶层await的语法报错问题。

这两个方案我都在自己的项目里验证过,既能在VS Code的Jupyter插件里正常分单元格运行,也能直接用python script.py在命令行执行,完全兼容两种场景。

内容的提问来源于stack exchange,提问作者Kerrick Staley

火山引擎 最新活动