Python 3.11中自定义Awaitable对象报错的最优重构方案

算力阿峰

2026-7-17

解决Python 3.11中asyncio.wait对自定义可等待对象的兼容性问题

问题原因

Python 3.11对asyncio.wait的底层实现做了调整：不再接受仅实现__await__方法的自定义可等待对象，要求传入的必须是Future/Task类的实例（这类对象自带add_done_callback方法）。Python 3.10的实现允许兼容这类自定义可等待对象，但3.11移除了该兼容逻辑，且未触发弃用警告，导致代码迁移后直接报错。

推荐的重构方式

方式一：用`asyncio.create_task`包装可等待对象（最直接）

无需修改自定义可等待类的实现，只需在调用asyncio.wait时，将每个自定义可等待对象用asyncio.create_task包装为Task实例即可：

import asyncio

class ExampleAwaitable:
    def __await__(self):
        async def _():
            return await asyncio.sleep(0)
        return _().__await__()

print(f"{asyncio.iscoroutine(ExampleAwaitable())=}")

async def amain():
    # 用create_task包装每个可等待对象
    await asyncio.wait([asyncio.create_task(ExampleAwaitable())])
    print("waited!")

asyncio.run(amain())

方式二：让自定义类继承`asyncio.Future`（更彻底）

如果你的自定义可等待对象需要支持Future的回调、结果设置等特性，可以直接继承asyncio.Future，并在内部实现异步逻辑：

import asyncio

class ExampleAwaitable(asyncio.Future):
    def __init__(self):
        super().__init__()
        # 启动内部异步任务，完成后设置结果
        asyncio.create_task(self._run())
    
    async def _run(self):
        await asyncio.sleep(0)
        self.set_result(None)

print(f"{asyncio.iscoroutine(ExampleAwaitable())=}")

async def amain():
    await asyncio.wait([ExampleAwaitable()])
    print("waited!")

asyncio.run(amain())