如何在Python中无需await即可启动或恢复异步协程？

Python协程本质是生成器，调用async函数仅创建协程对象，不会自动执行代码，必须通过await或事件循环调度才会运行。要模拟JS中async函数调用后立即执行至第一个await的行为，我们可以手动驱动协程执行到第一个等待点，再将后续逻辑交给事件循环后台处理。

核心实现思路 #

手动调用协程的send(None)方法驱动其执行，直到它返回第一个await对应的可等待对象（Future/协程等）。随后将该对象交给事件循环调度，通过回调链式处理后续执行步骤，最终返回一个Future用于后续获取结果。同时处理协程直接执行完毕、执行前抛异常等边界场景。

具体代码实现 #

import asyncio

def attempt(coro):
    """驱动协程执行至第一个await，后续逻辑交由事件循环后台运行"""
    loop = asyncio.get_running_loop()
    try:
        # 启动协程，执行到第一个await处
        pending = coro.send(None)
    except StopIteration as e:
        # 协程无await直接执行完毕，包装结果为已完成的Future
        final_fut = loop.create_future()
        final_fut.set_result(e.value)
        return final_fut
    except Exception as e:
        # 执行至第一个await前抛出异常，包装为已失败的Future
        final_fut = loop.create_future()
        final_fut.set_exception(e)
        return final_fut

    # 将await的对象统一转为可调度的Future/Task
    if asyncio.iscoroutine(pending):
        pending = loop.create_task(pending)
    elif not isinstance(pending, asyncio.Future):
        pending = asyncio.wrap_future(pending)

    # 回调函数：当前等待对象完成后，继续驱动协程执行
    def _resume_coro(fut):
        try:
            # 将当前等待结果发送给协程，继续执行到下一个await
            next_pending = coro.send(fut.result())
        except StopIteration as e:
            # 协程执行完毕，设置最终结果
            final_fut.set_result(e.value)
        except Exception as e:
            # 协程执行中抛出异常，传递异常
            final_fut.set_exception(e)
        else:
            # 递归处理下一个等待对象
            if asyncio.iscoroutine(next_pending):
                next_pending = loop.create_task(next_pending)
            elif not isinstance(next_pending, asyncio.Future):
                next_pending = asyncio.wrap_future(next_pending)
            next_pending.add_done_callback(_resume_coro)

    # 创建最终的Future，用于后续获取协程结果
    final_fut = loop.create_future()
    pending.add_done_callback(_resume_coro)
    return final_fut

使用示例 #

async def async_task(delay):
    print(f"Task {delay} starting")
    await asyncio.sleep(delay)
    print(f"Task {delay} finished")
    return f"Result: {delay}"

async def main():
    # 调用attempt后，协程立即执行到第一个await（打印starting）
    fut1 = attempt(async_task(2))
    fut2 = attempt(async_task(1))

    # 此时任务已在后台运行，可处理其他事务
    print("Doing other work...")
    await asyncio.sleep(0.5)
    print("Still doing other work...")

    # 最后等待任务结果
    res1 = await fut1
    res2 = await fut2
    print(res1, res2)

asyncio.run(main())

关键细节说明 #

• 手动驱动协程：通过coro.send(None)直接触发协程代码执行，跳过事件循环的调度等待，实现立即执行到第一个await的效果。
• 回调链式处理：每个await对象完成后，通过回调继续驱动协程执行，确保后续逻辑在后台自动运行。
• 异常全覆盖：处理了协程无await直接结束、执行前抛异常、await非标准可等待对象等场景，保证结果传递的可靠性。
• 兼容现有代码：无需修改原有async函数，仅用attempt包裹调用即可实现需求。

为什么之前的方案不可靠？ #

• asyncio.create_task + await asyncio.sleep(0)：依赖事件循环的调度优先级，sleep(0)仅让出控制权，但事件循环不一定会立即执行新创建的Task，批量启动时仍可能存在延迟。
• 装饰器方案：无法递归处理嵌套的async调用，内部协程仍需await才会执行，不符合预期行为。

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