如何子类化协程？实现MyCoro类与协程行为一致

要让MyCoro类的实例拥有和原生协程完全一致的行为，无需继承特定的Coroutine基类（Python标准库中没有直接可用的、能通过super().__init__直接封装协程的基类），只需让类的实例成为可等待对象即可，具体实现如下：

代码实现 #

import asyncio

class MyCoro:
    def __init__(self, text):
        async def coro():
            await asyncio.sleep(5)
            print(text)
        # 初始化时创建内部协程对象
        self._inner_coro = coro()
    
    def __await__(self):
        # 代理内部协程的await迭代器
        return self._inner_coro.__await__()

验证使用 #

你可以像原生协程一样直接将MyCoro实例传给asyncio.create_task：

async def main():
    # 与原生协程用法完全一致
    asyncio.create_task(MyCoro("Hello!"))
    await asyncio.sleep(6)  # 等待协程执行完成

asyncio.run(main())

原理说明 #

• asyncio.create_task接受可等待对象作为参数，包括原生协程、Task、Future，以及实现了__await__方法的自定义对象。
• 我们在MyCoro中实现__await__方法，直接返回内部协程对象的__await__迭代器，相当于让MyCoro实例完全复用原生协程的执行逻辑，因此行为和原生协程完全一致。

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