Python: оптимизация загрузки больших файлов с Requests и чанками

Для кого эта статья:

• Python-разработчики, работающие с большими данными
• Начинающие программисты, заинтересованные в оптимизации загрузок

• Специалисты, занимающиеся разработкой высоконагруженных систем и веб-приложений

  Работа с большими файлами в Python часто становится испытанием для разработчиков: неоптимизированный код может привести к переполнению памяти, таймаутам и даже падению всего приложения. Когда я впервые столкнулся с задачей загрузки датасетов размером в несколько гигабайт, стандартный вызов requests.get() превратил мою программу в пожирателя ОЗУ. К счастью, библиотека Requests предлагает элегантные решения для эффективной работы с крупными файлами — от потоковой загрузки до чтения по частям. Эти методы не просто экономят ресурсы, они кардинально меняют подход к получению данных из сети. 🚀

Осваивая Python и методы эффективной загрузки больших файлов, вы закладываете фундамент для разработки высоконагруженных систем. Курс Обучение Python-разработке от Skypro углубляет эти навыки, предлагая практические кейсы работы с Requests и другими библиотеками. Вы научитесь не только загружать гигабайтные файлы без перегрузки системы, но и создавать надежные веб-приложения, способные обрабатывать большие потоки данных в реальных проектах.

Проблемы при загрузке больших файлов в Python

Стандартный подход к загрузке файлов через библиотеку Requests выглядит обманчиво простым:

import requests

response = requests.get('https://example.com/largefile.zip')
with open('largefile.zip', 'wb') as f:
f.write(response.content)

Однако при работе с файлами размером в сотни мегабайт или гигабайты этот метод приводит к серьезным проблемам:

• Исчерпание оперативной памяти — весь файл загружается в RAM перед записью на диск
• Длительное время ожидания — пользователь не видит прогресс до полного завершения загрузки
• Потеря данных при прерывании — если соединение оборвется на 99% загрузки, весь процесс придется начинать заново
• Сложности с таймаутами — длительные загрузки могут превысить стандартные таймауты HTTP-соединений

Рассмотрим сравнение методов загрузки файлов и связанных с ними проблем:

Александр Петров, Lead Python Developer

Несколько лет назад нам нужно было загружать снимки со спутников для сервиса мониторинга сельхозугодий — файлы по 2-5 ГБ каждый. Первая версия системы использовала наивный подход с requests.get() и буквально "падала" каждый раз на крупных файлах. Диагностика показала, что Python-процесс потреблял всю доступную память и завершался с ошибкой OOM (Out of Memory).

После изучения документации Requests мы переписали загрузчик с использованием stream=True и обработки по чанкам. Потребление памяти упало с гигабайт до стабильных 10-15 МБ даже на самых больших файлах. Более того, добавив простую логику сохранения прогресса, мы смогли реализовать докачку прерванных загрузок — критически важную функцию для наших полевых станций с нестабильным интернетом.

Потоковая загрузка с Requests: метод stream=True

Первый шаг к оптимизации загрузки больших файлов — использование параметра stream=True при вызове метода get(). Этот простой флаг кардинально меняет поведение Requests:

• Соединение с сервером устанавливается сразу, но данные не загружаются в память автоматически
• Содержимое ответа становится доступным через итератор, позволяя обрабатывать его постепенно
• Разработчик получает контроль над процессом чтения и записи данных

Базовый пример потоковой загрузки выглядит так:

import requests

with requests.get('https://example.com/largefile.zip', stream=True) as response:
with open('largefile.zip', 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)

Ключевое отличие от стандартного подхода — использование метода iter_content(), который читает данные небольшими порциями (чанками), не загружая весь файл в память. Размер чанка (в примере 8192 байт) можно регулировать в зависимости от специфики задачи и доступных ресурсов.

Преимущества потокового подхода проявляются сразу в нескольких аспектах:

Однако необходимо учитывать некоторые особенности при работе с потоковой загрузкой:

1. Соединение с сервером остаётся открытым дольше, что может привести к таймаутам на некоторых серверах
2. Требуется явно закрывать соединение после использования (через контекстный менеджер или вызов response.close())
3. При использовании iter_content() без параметра decode_unicode=True необходимо самостоятельно обрабатывать кодировку, если требуется

Оптимизация памяти: скачивание файлов по частям

Хотя использование stream=True значительно улучшает управление памятью, для действительно больших файлов и систем с ограниченными ресурсами можно внедрить дополнительные оптимизации. Рассмотрим расширенные методы работы с частями файлов (чанками). 🧩

Оптимальный размер чанка зависит от нескольких факторов:

• Доступная память — чем меньше ОЗУ, тем меньше должен быть размер чанка
• Скорость сети — при быстром соединении больший размер чанка снижает накладные расходы
• Характер данных — для бинарных файлов размер чанка может быть любым, для текстовых важно не разделить символ

Вот более продвинутый пример с оптимизированной обработкой чанков:

import requests
import os

def download_file(url, filepath, chunk_size=8192):
"""
Оптимизированная загрузка файла с управлением памятью
"""
# Создаём папку, если не существует
os.makedirs(os.path.dirname(filepath), exist_ok=True)

# Получаем информацию о файле без скачивания
with requests.head(url, allow_redirects=True) as head:
file_size = int(head.headers.get('content-length', 0))

# Используем байтовый диапазон, если файл существует
downloaded = 0
headers = {}
if os.path.exists(filepath):
downloaded = os.path.getsize(filepath)
if downloaded < file_size:
# Продолжаем загрузку с нужной позиции
headers['Range'] = f'bytes={downloaded}-{file_size}'
elif downloaded == file_size:
print(f"Файл {filepath} уже скачан полностью")
return filepath

mode = 'ab' if downloaded > 0 else 'wb'

# Скачиваем файл по частям
with requests.get(url, stream=True, headers=headers) as response:
response.raise_for_status() # Проверяем статус ответа

with open(filepath, mode) as f:
for chunk in response.iter_content(chunk_size=chunk_size):
if chunk: # Фильтруем пустые чанки
f.write(chunk)
downloaded += len(chunk)

return filepath

Этот код содержит несколько важных оптимизаций:

1. Проверка размера файла перед загрузкой с помощью HEAD-запроса
2. Поддержка докачки частично загруженных файлов через HTTP Range headers
3. Фильтрация пустых чанков для экономии ресурсов
4. Контроль скачанных байтов для отслеживания прогресса

Мария Соколова, Data Engineer

Для проекта анализа геномных данных мне приходилось работать с файлами секвенирования ДНК размером 20-50 ГБ. Изначально я использовала обычную загрузку через requests, но столкнулась с постоянными сбоями из-за ограничений памяти на виртуальных машинах в облаке.

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

Эта técnica позволила нам сэкономить буквально недели времени — некоторые наборы данных требовали загрузки терабайт информации по нестабильным каналам связи. Более того, оптимизированный код мог работать на машинах всего с 2 ГБ RAM, что значительно снизило стоимость облачной инфраструктуры для проекта.

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

import tempfile
import shutil

def safe_download_huge_file(url, destination):
"""Безопасная загрузка огромных файлов с минимальным использованием памяти"""
# Создаем временный файл
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_path = temp_file.name

try:
# Загружаем во временный файл
with requests.get(url, stream=True) as response:
with open(temp_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024*1024): # 1MB chunks
if chunk:
f.write(chunk)
# Сбрасываем буфер на диск
f.flush()
os.fsync(f.fileno())

# Перемещаем в конечное место назначения
shutil.move(temp_path, destination)
return destination
except Exception as e:
# Удаляем временный файл при ошибке
if os.path.exists(temp_path):
os.unlink(temp_path)
raise e

Отслеживание прогресса загрузки больших файлов

Отслеживание прогресса загрузки — это не просто улучшение пользовательского опыта, но и важный инструмент для диагностики проблем при работе с большими файлами. Реализация индикатора загрузки позволяет:

• Информировать пользователя о примерном времени ожидания
• Выявлять проблемы с сетевым соединением (если скорость загрузки резко падает)
• Контролировать правильность получения всего контента
• Обеспечивать возможность отмены длительных загрузок

Простейшая реализация отслеживания прогресса выглядит так:

import requests
import sys
from tqdm import tqdm # pip install tqdm

def download_with_progress(url, filename):
response = requests.get(url, stream=True)
# Получаем размер файла из заголовков
total_size = int(response.headers.get('content-length', 0))

# Создаем прогресс-бар
with tqdm(total=total_size, unit='B', unit_scale=True, desc=filename) as pbar:
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024):
if chunk:
f.write(chunk)
# Обновляем прогресс-бар
pbar.update(len(chunk))

return filename

Библиотека tqdm обеспечивает элегантный прогресс-бар в консоли, но можно реализовать и собственное решение для отслеживания:

def download_with_custom_progress(url, filename):
response = requests.get(url, stream=True)
total_size = int(response.headers.get('content-length', 0))
downloaded = 0

# Размер чанка: 1MB
chunk_size = 1024 * 1024

with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)

# Расчет процента загрузки
percent = int(100 * downloaded / total_size) if total_size > 0 else 0

# Вывод прогресса в консоль
sys.stdout.write(f"\rЗагружено {downloaded/(1024*1024):.2f} MB из {total_size/(1024*1024):.2f} MB ({percent}%)")
sys.stdout.flush()

print() # Перевод строки после завершения
return filename

Для более сложных сценариев можно реализовать расчёт скорости загрузки и оценку времени до завершения:

import time

def download_with_eta(url, filename):
response = requests.get(url, stream=True)
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
start_time = time.time()
chunk_size = 1024 * 1024 # 1MB

with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)

# Рассчитываем скорость и ETA
elapsed_time = time.time() – start_time
if elapsed_time > 0:
speed = downloaded / elapsed_time
eta = (total_size – downloaded) / speed if speed > 0 else 0

# Форматируем время в читаемом виде
if eta < 60:
eta_str = f"{eta:.0f} сек"
elif eta < 3600:
eta_str = f"{eta/60:.1f} мин"
else:
eta_str = f"{eta/3600:.1f} часов"

percent = int(100 * downloaded / total_size) if total_size > 0 else 0
sys.stdout.write(
f"\r{percent}% | {downloaded/(1024*1024):.1f}/{total_size/(1024*1024):.1f} MB | "
f"{speed/(1024*1024):.1f} MB/s | Осталось: {eta_str}"
)
sys.stdout.flush()

print(f"\nЗагрузка завершена за {time.time() – start_time:.1f} секунд")
return filename

Практические кейсы использования Requests для скачивания

Теория важна, но настоящее мастерство приходит с практикой. Рассмотрим несколько реальных сценариев, где оптимизированные методы загрузки с Requests решают конкретные задачи. 📊

1. Параллельная загрузка нескольких файлов

Когда необходимо загрузить множество файлов, последовательный подход может быть неэффективным. Библиотека concurrent.futures позволяет организовать многопоточную загрузку:

import concurrent.futures
import requests
import time

def download_file(url, filename):
with requests.get(url, stream=True) as response:
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
return filename

def download_multiple_files(urls_and_filenames):
start_time = time.time()
# Определяем оптимальное количество потоков
max_workers = min(32, len(urls_and_filenames))

with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Запускаем загрузку в нескольких потоках
future_to_url = {
executor.submit(download_file, url, filename): (url, filename)
for url, filename in urls_and_filenames
}

# Обрабатываем результаты по мере завершения
for future in concurrent.futures.as_completed(future_to_url):
url, filename = future_to_url[future]
try:
downloaded_file = future.result()
print(f"Загружен файл: {downloaded_file}")
except Exception as e:
print(f"Ошибка при загрузке {url}: {e}")

print(f"Общее время загрузки: {time.time() – start_time:.2f} секунд")

2. Загрузка файлов через прокси и с аутентификацией

В корпоративных средах часто требуется загружать файлы через прокси-серверы или с использованием аутентификации:

def download_through_proxy(url, filename, proxy_url, auth=None):
proxies = {
'http': proxy_url,
'https': proxy_url
}

with requests.get(
url, 
stream=True, 
proxies=proxies,
auth=auth, # Basic Auth (username, password)
verify=False # Отключить проверку SSL, если требуется
) as response:
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)

return filename

3. Загрузка файла с возобновлением

Полное решение для надежной загрузки с возможностью возобновления прерванных скачиваний:

def resumable_download(url, filename, max_retries=5, retry_delay=5):
"""
Надежная загрузка файла с поддержкой возобновления
и автоматическими повторными попытками.
"""
headers = {}
downloaded_bytes = 0

if os.path.exists(filename):
downloaded_bytes = os.path.getsize(filename)
headers['Range'] = f'bytes={downloaded_bytes}-'

retries = 0
while retries < max_retries:
try:
with requests.get(url, headers=headers, stream=True) as response:
# Проверяем, поддерживает ли сервер возобновление
if downloaded_bytes > 0 and response.status_code != 206:
# Сервер не поддерживает докачку
print("Сервер не поддерживает возобновление загрузки. Начинаем заново.")
os.remove(filename)
headers = {}
downloaded_bytes = 0
continue

response.raise_for_status()
total_size = int(response.headers.get('content-length', 0)) + downloaded_bytes

mode = 'ab' if downloaded_bytes > 0 else 'wb'
with open(filename, mode) as f:
for chunk in response.iter_content(chunk_size=1024*1024):
if chunk:
f.write(chunk)
downloaded_bytes += len(chunk)
# Показываем прогресс
percent = int(100 * downloaded_bytes / total_size) if total_size > 0 else 0
print(f"\rЗагружено: {percent}% ({downloaded_bytes/(1024*1024):.1f} MB)", end='')

print("\nЗагрузка завершена успешно!")
return True

except (requests.exceptions.RequestException, IOError) as e:
retries += 1
print(f"\nОшибка: {str(e)}. Повторная попытка {retries}/{max_retries} через {retry_delay} секунд...")
time.sleep(retry_delay)
# Увеличиваем время задержки для следующей попытки
retry_delay *= 1.5

print(f"Не удалось загрузить файл после {max_retries} попыток")
return False

Сравнение производительности разных методов загрузки:

Примечательно, что использование слишком маленьких чанков может немного увеличить общее время загрузки из-за дополнительных накладных расходов на обработку каждого чанка, но радикально снижает потребление памяти. Многопоточная загрузка значительно ускоряет процесс, но требует более сложной реализации и контроля ресурсов.

Работа с большими файлами в Python — это искусство балансирования между производительностью, надежностью и эффективным использованием ресурсов. Библиотека Requests предоставляет все необходимые инструменты: от потоковой загрузки до управления соединениями. Выбирая правильный метод для своей задачи, вы можете скачивать гигабайтные файлы даже на ограниченном оборудовании, сохраняя контроль над каждым аспектом процесса. Помните: правильная обработка больших данных — это фундаментальный навык, отличающий профессионального разработчика от новичка.