Async GET request for images using aiohttp

async_download_images.py

import re
import asyncio
from io import BytesIO
import time
from pathlib import Path
from PIL import Image

import aiohttp
import requests


async def save_image(session, image_url: str, save_path: Path):
    async with session.get(image_url) as resp:
        content = await resp.content.read()
        Image.open(BytesIO(content)).save(save_path)
        print(f"Saved {save_path}")


async def _download_images(image_urls: list[str], download_location: Path | str):
    async with aiohttp.ClientSession(headers={"Connection": "keep-alive"}) as session:
        tasks = []
        for image_url in image_urls:
            image_name = image_url.split("/")[-1]
            extension = image_name.split(".")[-1]
            if not image_name.endswith(extension):
                raise ValueError(image_name)
            download_path = Path(download_location) / image_name
            tasks.append(
                asyncio.ensure_future(save_image(session, image_url, download_path))
            )
        await asyncio.gather(*tasks)


def download_images(image_urls: list[str], download_location: Path | str):
    start_time = time.time()
    asyncio.run(_download_images(image_urls, download_location))
    elapsed = time.time() - start_time
    print(f"--- {elapsed:.2f} seconds toal ---")
    print(f"--- {len(image_urls) / elapsed:.2f} images per second ---")


if __name__ == "__main__":
    img_urls = [
        f"https://raw.githubusercontent.com/PokeAPI/sprites/master/sprites/pokemon/{n}.png"
        for n in range(258, 300)
    ]

    save_dir = Path(".tmp/images")
    save_dir.mkdir(exist_ok=True)
    download_images(
        img_urls,
        save_dir,
    )

adapted from https://www.twilio.com/blog/asynchronous-http-requests-in-python-with-aiohttp

very useful, thanks 🫶🏼