如何用装饰器将requests图片下载函数转为异步请求回调函数？

我现在有一个同步下载图片的代码，使用requests库遍历URL列表下载到指定目录：

import pathlib
import requests

def get_image(url, image_name):
    path = pathlib.Path('/path/to/some/directory')
    response = requests.get(url, stream=True)
    with open('{}/{}.png'.format(path, image_name), 'wb') as file:
        for block in response.iter_content(1024):
            file.write(block)

# 使用方式
for url in urls:
    get_image(url, image_name)

我想知道是否可以创建一个装饰器，将这个get_image函数转为requests异步请求的回调函数？

当然可以！不过先说明一点：requests本身是同步HTTP库，没有原生的异步API，所以如果要实现高效的异步下载，更推荐搭配aiohttp这个专门的异步HTTP工具来做。下面我会给出两种装饰器方案，帮你把现有的同步函数适配成异步流程的一部分：

方案1：装饰器包装为异步请求回调（基于aiohttp） #

首先安装aiohttp：

pip install aiohttp

这个装饰器会帮你把同步的图片保存逻辑，和异步HTTP请求结合起来，让整个下载流程变成异步的：

import asyncio
import aiohttp
import pathlib
from functools import wraps

def async_download_callback(func):
    """装饰器：将同步图片保存函数转为异步下载的回调逻辑"""
    @wraps(func)
    async def wrapper(url, image_name):
        # 用aiohttp发起异步HTTP请求
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                # 异步读取图片内容
                content = await response.read()
                # 把同步的保存操作放到线程池执行，避免阻塞事件循环
                loop = asyncio.get_event_loop()
                await loop.run_in_executor(None, func, url, image_name, content)
    return wrapper

# 调整原有的get_image函数，支持直接传入内容（兼容原有同步逻辑）
def get_image(url, image_name, content=None):
    save_path = pathlib.Path('/path/to/some/directory')
    save_path.mkdir(exist_ok=True)  # 确保目录存在
    file_path = save_path / f"{image_name}.png"
    
    if content is not None:
        # 用传入的内容直接保存（异步流程走这里）
        with open(file_path, 'wb') as f:
            f.write(content)
    else:
        # 原有同步请求逻辑（保留兼容）
        response = requests.get(url, stream=True)
        with open(file_path, 'wb') as f:
            for block in response.iter_content(1024):
                f.write(block)

# 用装饰器得到异步版本的下载函数
async_get_image = async_download_callback(get_image)

# 批量异步下载的入口函数
async def batch_download(url_list):
    tasks = []
    for idx, url in enumerate(url_list):
        # 为每个URL创建异步任务
        task = asyncio.create_task(async_get_image(url, f"image_{idx}"))
        tasks.append(task)
    # 等待所有任务完成
    await asyncio.gather(*tasks)

# 运行异步程序
if __name__ == "__main__":
    sample_urls = [
        "https://example.com/image1.png",
        "https://example.com/image2.png"
    ]
    asyncio.run(batch_download(sample_urls))

方案2：通用装饰器将同步函数转为异步函数 #

如果你只是想让原有的get_image能在异步环境中被调用，不需要直接绑定请求回调，可以用这个通用装饰器把同步函数包装成异步函数：

import asyncio
import pathlib
import requests
from functools import wraps

def sync_to_async(func):
    """通用装饰器：把同步函数转为可在异步环境中调用的函数"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 把同步函数放到线程池执行，避免阻塞异步事件循环
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, func, *args, **kwargs)
    return wrapper

# 直接用装饰器包装原有的get_image
@sync_to_async
def get_image(url, image_name):
    save_path = pathlib.Path('/path/to/some/directory')
    save_path.mkdir(exist_ok=True)
    response = requests.get(url, stream=True)
    with open(save_path / f"{image_name}.png", 'wb') as f:
        for block in response.iter_content(1024):
            f.write(block)

# 批量异步下载
async def batch_download(url_list):
    # 创建所有异步任务并等待完成
    tasks = [get_image(url, f"image_{idx}") for idx, url in enumerate(url_list)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    sample_urls = [
        "https://example.com/image1.png",
        "https://example.com/image2.png"
    ]
    asyncio.run(batch_download(sample_urls))

关键说明： #

• 因为requests是同步库，直接在异步事件循环里调用会阻塞整个循环，所以两种方案都用了run_in_executor把同步IO操作放到线程池执行，保证异步并发的效率。
• 如果追求最优的异步性能，优先选方案1，用aiohttp发起真正的异步HTTP请求，而不是在线程池里跑requests。

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