Python3中如何创建自定义asyncio可await对象？

Python中可被await的对象有三类：协程对象、实现了__await__方法的实例、asyncio内置的Future/Task对象，以下分别提供两种常用的封装urllib.request的实现方案：

方式1：async函数封装（最常用，开发优先选择） #

async函数返回的协程本身就是合法可等待对象，和asyncio.sleep的使用体验完全一致。由于urllib.request是同步阻塞IO，需要配合线程池执行避免阻塞asyncio事件循环：

import asyncio
import urllib.request
import time

# 自定义可等待的HTTP请求方法
async def async_http_get(url: str, timeout: int = 10) -> str:
    # 同步执行的urllib请求逻辑
    def sync_request():
        with urllib.request.urlopen(url, timeout=timeout) as resp:
            return resp.read().decode("utf-8")
    
    # 将同步逻辑扔到线程池执行，等待过程可挂起不卡事件循环
    return await asyncio.to_thread(sync_request)

async def do_something(number: int, url: str):
    print(f"do something no.{number} 开始时间: {time.perf_counter():6.2f}")
    # 直接await自定义的可等待逻辑
    content = await async_http_get(url)
    print(f"do something no.{number} 结束时间: {time.perf_counter():6.2f}，响应长度: {len(content)}")

async def main():
    # 可同时并发多个请求，不会互相阻塞
    await asyncio.gather(
        do_something(1, "https://www.example.com"),
        do_something(2, "https://www.example.org")
    )

if __name__ == "__main__":
    asyncio.run(main())

方式2：自定义类实现__await__方法（适合复杂封装场景） #

如果需要做更复杂的逻辑扩展（比如内置重试、缓存、回调钩子等），可以自己实现带__await__方法的类，实例本身就是可等待对象：

import asyncio
import urllib.request
import time
from collections.abc import Awaitable

class AsyncHttpRequest(Awaitable):
    def __init__(self, url: str, timeout: int = 10, retry: int = 2):
        self.url = url
        self.timeout = timeout
        self.retry = retry # 可自定义扩展参数，比如重试次数
        self._loop = asyncio.get_running_loop()
    
    def _sync_request_task(self):
        # 这里可以自定义重试、异常处理等逻辑
        for i in range(self.retry + 1):
            try:
                with urllib.request.urlopen(self.url, timeout=self.timeout) as resp:
                    return resp.read().decode("utf-8")
            except Exception as e:
                if i == self.retry:
                    raise e
    
    def __await__(self):
        future = self._loop.run_in_executor(None, self._sync_request_task)
        return (yield from future.__await__())

# 测试代码
async def do_something(number: int, url: str):
    print(f"do something no.{number} 开始时间: {time.perf_counter():6.2f}")
    # 直接await类实例，和asyncio.sleep使用方式完全一致
    content = await AsyncHttpRequest(url, retry=3)
    print(f"do something no.{number} 结束时间: {time.perf_counter():6.2f}，响应长度: {len(content)}")

async def main():
    await asyncio.gather(
        do_something(1, "https://www.example.com"),
        do_something(2, "https://www.example.org")
    )

if __name__ == "__main__":
    asyncio.run(main())

注意事项 #

• 所有同步IO逻辑必须放到线程/进程池执行，不能直接在事件循环主线程调用urllib.request，否则会阻塞所有异步任务，失去异步的优势
• Python 3.9以下版本没有asyncio.to_thread，可以换成loop.run_in_executor(None, sync_request)实现相同效果

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