Пишем асинхронный загрузчик файлов на python

Иногда бывает полезно вставить куча ссылок и скачать их все разом, а не по очереди, особенно если при этом можно смотреть прогресс бар наших загрузок. Реализуем подобную тулу на python.

requirements.txt

Plaintext

asyncio==3.4.3
httpx==0.24.1
requests==2.31.0
tqdm==4.65.0

Это либы, которые нам понадобятся. Также хочется иметь возможность переопределить путь до скачанных файлов, поэтому напишем cli (command line interface):

Python

import argparse
import os
from dataclasses import dataclass

@dataclass
class CommandLineParser:
    """Основной класс для CLI"""
    urls_file_path: str
    save_dir: str

    @classmethod
    def create_parser(cls):
        """Создаем CLI в методе класса"""
        parser = argparse.ArgumentParser(description="Async downloader")
        parser.add_argument(
            "-u", "--urls", dest="urls_file_path",
            required=False,
            default="urls.json",
            help="Path for url file"
        )
        parser.add_argument(
            "-d", "--dir",
            dest="save_dir",
            required=False,
            default=os.getcwd(),
            help="Save directory for download files"
        )
        
        return parser

Наш класс будет принимать всего два аргумента, это путь до файла с урлами(ссылки на скачивание) и директорию для сохранения. По дефолту оба параметра в CLI являются необязательными и имеют значения по умолчанию.

Создадим файл urls.json:

JSON

[
    ["http://ipv4.download.thinkbroadband.com/50MB.zip", "50MB.zip"],
    ["http://ipv4.download.thinkbroadband.com/10MB.zip", "10MB.zip"],
    ["http://ipv4.download.thinkbroadband.com/20MB.zip", "20MB.zip"]
]

Тут мы используем тестовый сайт для закачки файлов. Вторым элементом в массиве идет имя файла.

Основной код:

Python

import asyncio
import httpx
import tqdm
import json
import os
from cli import CommandLineParser

args = CommandLineParser.create_parser().parse_args()
cmd_args = CommandLineParser(**vars(args))

def read_json_file(file_path):
    with open(file_path, 'r') as file:
        loaded_urls = json.load(file)
    return loaded_urls


async def download_files(url: str, filename: str, save_path: str):
    full_save_path = os.path.join(save_path, filename)

    if not os.path.exists(save_path):
        os.makedirs(save_path)

    with open(full_save_path, 'wb') as f:
        async with httpx.AsyncClient() as client:
            async with client.stream('GET', url) as r:
                
                r.raise_for_status()
                total = int(r.headers.get('content-length', 0))

                tqdm_params = {
                    'desc': url,
                    'total': total,
                    'miniters': 1,
                    'unit': 'it',
                    'unit_scale': True,
                    'unit_divisor': 1024,
                }

                with tqdm.tqdm(**tqdm_params) as pb:
                    async for chunk in r.aiter_bytes():
                        pb.update(len(chunk))
                        f.write(chunk)


async def main():
    loop = asyncio.get_running_loop()

    urls = read_json_file(cmd_args.urls_file_path)

    tasks = [loop.create_task(download_files(url, filename, cmd_args.save_dir)) for url, filename in urls]
    await asyncio.gather(*tasks, return_exceptions=True)


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

Метод read_json_file(file_path) — читаем наш json файл.

async def download_files(url: str, filename: str, save_path: str) — асинхронная функция загрузки файлов. Внутри определяем полный путь до файла и создаем директорию для загрузки, если она не создана. Далее через async отправляем запросы и там же подключаем наш прогресс бар с набором параметров.

async def main() — основной метод из которого стартует остальное.

Пример работы: