寻求适配Jupyter Notebook的Python asyncio安全运行封装方案

针对Python 3.8环境下同时兼容普通Python环境与Jupyter Notebook的异步协程同步封装需求，以下是无需依赖nest_asyncio、无线程、无私有API的安全实现：

import asyncio
from typing import Awaitable, Any

ASYNC_IO_NO_RUNNING_LOOP_MSG = 'no running event loop'

def jupyter_safe_run_coroutine(async_coroutine: Awaitable[Any], _test_mode: bool = False) -> Any:
    try:
        # 尝试获取当前运行中的事件循环
        loop = asyncio.get_running_loop()
        # 将协程提交到运行中的loop，返回可同步等待的Future
        future = asyncio.run_coroutine_threadsafe(async_coroutine, loop)
        # 同步等待任务完成并返回结果
        return future.result()
    except RuntimeError as e:
        if _test_mode:
            raise e
        # 无运行中的事件循环，使用asyncio.run直接启动
        if ASYNC_IO_NO_RUNNING_LOOP_MSG in str(e):
            return asyncio.run(async_coroutine)
    except Exception as e:
        # 透传其他异常
        raise e

方案说明 #

1. 环境适配逻辑

   • 先尝试获取当前运行中的事件循环：若成功（如Jupyter Notebook环境），则使用asyncio.run_coroutine_threadsafe将协程提交到现有loop，通过future.result()同步等待任务完成。
   • 若获取loop失败（普通Python环境，无运行中的loop），则直接使用asyncio.run()启动协程，这也是标准的异步启动方式。

2. 为什么不用nest_asyncio

   • 避免任务饥饿问题：任务由Jupyter原生loop调度，遵循正常的事件循环优先级，不会出现任务抢占异常。
   • 解决日志输出问题：协程中的日志会通过原生loop正常输出到Jupyter控制台，不会丢失。
   • 消除内核崩溃风险：无需patch事件循环的底层逻辑，避免了对Jupyter内核稳定性的影响。

3. 兼容性与限制满足

   • 兼容Python 3.8：asyncio.run_coroutine_threadsafe在Python 3.7及以上版本可用。
   • 无线程依赖：虽然方法名包含threadsafe，但它仅用于保证跨线程提交任务的安全性，在当前线程（Jupyter内核线程）中使用完全安全，无需额外创建线程。
   • 无需asyncio.Runner：符合Python 3.8的环境限制。

使用示例 #

用户端调用方式保持不变，完全无需感知异步实现：

import my_lib
import pandas as pd

df = pd.DataFrame(data='some data')
enriched_df = my_lib.enrich(df)

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