如何让含asyncio的代码同时兼容Jupyter笔记本与脚本运行？

我习惯用带# %%单元格分隔符的普通.py文件来创建Jupyter笔记本，示例如下：

# %%
import polars as pl
import numpy as np
# %%
df = pl.read_parquet('my_cool.parquet')
...

这类文件既能通过VS Code的Jupyter插件作为笔记本运行，也能直接在命令行里当作脚本执行。

但当我在其中使用asyncio时遇到了冲突：Jupyter依赖autoawait特性，要求在顶层代码直接使用await；而脚本运行时必须在顶层调用asyncio.run()，两种写法无法兼容：

# %%
# 这段代码在Jupyter中能正常运行，但作为脚本执行会报错：
# SyntaxError: 'await' outside function
foo = await my_async_func()
# %%
# 这段代码作为脚本能正常运行，但在Jupyter中会报错：
# RuntimeError: asyncio.run() cannot be called from a running event loop
bar = asyncio.run(my_async_func())

我试过以下方法，但都无效：

• nest-asyncio：和我使用的部分库不兼容
• 在导入前执行get_ipython().autoawait = False：无法解决Jupyter中的问题
• 使用shebang命令#!/usr/bin/env -S uv run ipython：无法修复脚本运行的问题

请问该如何编写能在两种环境下都正常运行的代码？

核心思路是检测当前运行环境，自动切换适配的异步执行方式，避免两种环境的冲突：

方法1：基础环境检测 #

直接判断是否处于IPython/Jupyter环境，分支执行对应逻辑：

import asyncio

async def my_async_func():
    # 示例异步函数
    await asyncio.sleep(1)
    return "完成"

# 核心适配逻辑
try:
    # 尝试获取IPython对象，判断是否在Jupyter环境
    get_ipython()
    # Jupyter环境下直接用await
    result = await my_async_func()
except NameError:
    # 非Jupyter环境，用asyncio.run()
    result = asyncio.run(my_async_func())

print(result)

方法2：严谨环境校验 #

如果担心get_ipython()在非Jupyter的IPython环境中存在，可以进一步校验内核信息：

import asyncio

async def my_async_func():
    await asyncio.sleep(1)
    return "完成"

def is_jupyter_env():
    try:
        from IPython import get_ipython
        ip = get_ipython()
        # 检查是否为Jupyter内核运行
        return "IPKernelApp" in ip.config
    except (ImportError, AttributeError):
        return False

# 根据环境执行
if is_jupyter_env():
    result = await my_async_func()
else:
    result = asyncio.run(my_async_func())

print(result)

方法3：封装通用执行工具 #

如果有多个异步函数需要调用，封装一个工具函数可以简化代码：

import asyncio

def run_async(coro):
    try:
        # 检测Jupyter环境
        get_ipython()
        return coro
    except NameError:
        return asyncio.run(coro)

# 使用示例
async def my_async_func():
    await asyncio.sleep(1)
    return "完成"

# 两种环境下都能正常运行
result = await run_async(my_async_func())
# 脚本环境下也可直接调用：result = run_async(my_async_func())
print(result)

这些方案通过环境判断实现分支执行，既满足Jupyter的autoawait要求，又避免了脚本环境的语法错误，同时不会触发Jupyter中asyncio.run()的事件循环冲突。

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