如何在asyncio.run的运行环境中正常调用curio.spawn？

问题根源 #

你遇到的报错本质是curio和asyncio是两个互不兼容的Python异步运行时，curio的API依赖自身专属的内核调度逻辑，直接在asyncio的事件循环中调用curio.spawn这类原生API时，asyncio无法识别curio内部的调度请求，就会抛出RuntimeError: Task got bad yield错误。

可落地的解决方法 #

方案1：使用curio官方asyncio兼容桥

curio内置了curio.asyncio模块，支持将curio协程封装成asyncio可直接等待的对象，不需要修改原有业务逻辑，适配成本最低。
适配示例代码：

import curio
import asyncio
from curio.asyncio import aio_as_coroutine

async def curio_method():
    await curio.sleep(1)
    print("curio 方法执行完成")

# 将curio逻辑封装为asyncio可调用的协程
@aio_as_coroutine
async def wrap_curio_logic():
    task = await curio.spawn(curio_method())
    await task

async def asyncio_method():
    # 直接等待封装后的协程即可
    await wrap_curio_logic()

asyncio.run(asyncio_method())

方案2：猴子补丁替换curio.spawn实现

如果你不需要用到curio的专属特性，只需要让第三方库的spawn逻辑适配asyncio，可以直接在项目初始化阶段替换curio.spawn的实现，用asyncio原生的任务调度逻辑覆盖：
补丁示例代码：

import curio
import asyncio

# 项目初始化阶段提前打补丁
def patch_curio_spawn():
    async def asyncio_spawn(coro, *args, **kwargs):
        # 直接用asyncio.create_task代替curio的spawn逻辑
        return asyncio.create_task(coro)
    curio.spawn = asyncio_spawn

patch_curio_spawn()

# 之后原有代码不需要修改就能正常运行
async def curio_method():
    await asyncio.sleep(1)
    print("curio 方法执行完成")

async def asyncio_method():
    task = await curio.spawn(curio_method())
    await task

asyncio.run(asyncio_method())

注意：该方案需要确认第三方库用到的curio API只有spawn相关逻辑，如果用到了更多curio专属API还需要补充对应补丁

方案3：独立线程运行curio内核

如果上述两个方案都不适用，可以单独启动一个线程跑curio的内核，asyncio和curio两个运行时之间用线程安全的队列传递消息，完全隔离两个运行时的调度逻辑。

内容的提问来源于stack exchange，提问作者Gary van der Merwe