Загрузка файлов — вроде бы скучная задача. Но когда у тебя десятки или сотни URL-ов, а скорость важна — тут обычный requests пасует. На помощь приходит асинхронность: asyncio, aiohttp, очереди, семафоры — и твой загрузчик превращается в реактивный снаряд.
Сегодня мы напишем асинхронный downloader, который умеет:
- качать кучу файлов одновременно (и не убивать интернет)
- уважать лимиты (через семафоры)
- обрабатывать ошибки (в духе "скачать не удалось — но мы не падаем")
- сохранять файлы правильно
- работать с задачами из очереди
⚙️ Зависимости
pip install aiohttp aiofiles
- aiohttp — асинхронный HTTP-клиент
- aiofiles — асинхронная работа с файлами
🧱 Пример 1: Базовый async downloader
import aiohttp
import aiofiles
import asyncio
async def download_file(session, url, filename):
try:
async with session.get(url) as response:
response.raise_for_status() # выбросим исключение, если код ошибки
async with aiofiles.open(filename, 'wb') as f:
content = await response.read()
await f.write(content)
print(f"✅ Скачано: {filename}")
except Exception as e:
print(f"❌ Ошибка при скачивании {url}: {e}")
🧑🔬 Комментарии:
- session.get(url) — асинхронный HTTP-запрос
- aiofiles.open — неблокирующее открытие файла
- response.read() — читаем весь контент
- Всё в try-except, чтобы не свалиться при ошибке
🎯 Пример 2: Загрузка нескольких файлов параллельно
async def main():
urls = [
("https://example.com/file1.jpg", "file1.jpg"),
("https://example.com/file2.jpg", "file2.jpg"),
# Добавь свои URL
]
async with aiohttp.ClientSession() as session:
tasks = [download_file(session, url, filename) for url, filename in urls]
await asyncio.gather(*tasks)
asyncio.run(main())
🧑🔬 Комментарии:
- asyncio.gather(*tasks) — выполняем задачи параллельно
- ClientSession создаёт пул соединений (очень желательно)
🕳️ Пример 3: Ограничим количество одновременных загрузок (через семафор)
Если качать 1000 файлов за раз — можно попасть в бан. Используем семафор:
semaphore = asyncio.Semaphore(5) # максимум 5 одновременных загрузок
async def limited_download(session, url, filename):
async with semaphore:
await download_file(session, url, filename)
Теперь поменяем в main:
tasks = [limited_download(session, url, filename) for url, filename in urls]
📦 Пример 4: Очередь задач (asyncio.Queue)
Полезно, если задачи поступают динамически или мы хотим более контролируемое выполнение.
async def worker(name, queue, session):
while True:
url, filename = await queue.get()
await download_file(session, url, filename)
queue.task_done()
print(f"🔁 {name} завершил работу над {filename}")
async def main_with_queue():
queue = asyncio.Queue()
urls = [
("https://example.com/file1.jpg", "file1.jpg"),
("https://example.com/file2.jpg", "file2.jpg"),
# ...
]
for item in urls:
await queue.put(item)
async with aiohttp.ClientSession() as session:
workers = [asyncio.create_task(worker(f"Worker-{i}", queue, session)) for i in range(5)]
await queue.join()
for w in workers:
w.cancel()
asyncio.run(main_with_queue())
🧑🔬 Комментарии:
- queue.get() — берём задачу
- queue.task_done() — отмечаем, что задача завершена
- queue.join() — ждём, пока все задачи из очереди не будут выполнены
🚫 Пример 5: Обработка неудачных загрузок + повтор
Добавим повтор, если загрузка упала:
async def download_with_retries(session, url, filename, retries=3):
for attempt in range(1, retries + 1):
try:
await download_file(session, url, filename)
return
except Exception as e:
print(f"⚠️ Попытка {attempt} не удалась для {filename}")
await asyncio.sleep(1)
print(f"❌ Все попытки провалились для {filename}")
И используем вместо download_file в worker.
🎉 Бонус: Автоматическое имя файла из URL
from urllib.parse import urlparse
import os
def get_filename_from_url(url):
return os.path.basename(urlparse(url).path) or "downloaded_file"
Дополним наш асинхронный загрузчик тремя новыми фишками для настоящего продакшн-качества:
📜 1. Логгирование в файл (с помощью logging)
Заменим print на продвинутый логгер, чтобы всё записывалось и в консоль, и в файл:
import logging
# Настройка логгера
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)s | %(message)s',
handlers=[
logging.FileHandler("downloader.log"),
logging.StreamHandler()
]
)
# Использование
logging.info("Начинаем загрузку...")
logging.warning("Это предупреждение")
logging.error("Ошибка при загрузке файла")
В функциях меняем:
print(f"✅ Скачано: {filename}")
# →
logging.info(f"✅ Скачано: {filename}")
📊 2. Прогресс-бар с tqdm
asyncio и tqdm не очень дружат в лоб, но у нас есть решение.
Установим:
pip install tqdm
Добавим прогресс-бар через tqdm.asyncio или просто в main:
from tqdm.asyncio import tqdm_asyncio
from tqdm import tqdm
async def main():
urls = [
("https://example.com/file1.jpg", "file1.jpg"),
("https://example.com/file2.jpg", "file2.jpg"),
# и так далее
]
async with aiohttp.ClientSession() as session:
tasks = [
limited_download(session, url, filename)
for url, filename in urls
]
# Оборачиваем с tqdm
for f in tqdm_asyncio.as_completed(tasks, desc="📥 Загрузка файлов", total=len(tasks)):
await f
Или попроще:
for f in tqdm(tasks):
await f
🖼️ 3. GUI с PyQt5 (или tkinter, кому что милее)
Покажу минимальный пример с PyQt5:
pip install PyQt5
from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QTextEdit
import sys
class DownloaderApp(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("Async Downloader")
self.setGeometry(100, 100, 400, 300)
layout = QVBoxLayout()
self.log_output = QTextEdit()
self.log_output.setReadOnly(True)
self.download_button = QPushButton("Начать загрузку")
self.download_button.clicked.connect(self.start_download)
layout.addWidget(self.log_output)
layout.addWidget(self.download_button)
self.setLayout(layout)
def log(self, message):
self.log_output.append(message)
def start_download(self):
# Запуск асинхронной загрузки здесь (можно через asyncio + QEventLoop)
self.log("🚀 Начинаем загрузку...")
app = QApplication(sys.argv)
window = DownloaderApp()
window.show()
sys.exit(app.exec_())
🔌 Интеграция с асинхронностью
Для полноценной работы нужно подружить asyncio с Qt. Используем qasync:
pip install qasync
А дальше всё работает через QApplication + asyncio.
🧩 Заключение
Теперь у нас:
- ✔️ Асинхронный загрузчик
- ✔️ Ограничения через семафоры
- ✔️ Очередь задач
- ✔️ Логгирование в файл и консоль
- ✔️ Прогресс-бар
- ✔️ GUI-обёртка на PyQt