Загрузка файлов через HTTP-запросы в Python

Скачивание с использованием сессии для сохранения кук

Объект Session позволяет переносить куки и заголовки между запросами, что полезно при загрузке файлов с сайтов, требующих авторизации.

import requests

with requests.Session() as session:
    # Авторизация (пример с Basic Auth)
    session.auth = ("user", "pass")
    # Или через форму
    # session.post("https://example.com/login", data={"login": "u", "password": "p"})
    
    url = "https://example.com/protected/file.txt"
    response = session.get(url, stream=True)
    response.raise_for_status()
    
    with open("file.txt", "wb") as f:
        for chunk in response.iter_content(8192):
            if chunk:
                f.write(chunk)

# Результат: файл file.txt успешно скачан из защищённого раздела

Сессия автоматически обрабатывает редиректы и сохраняет куки после логина.

Скачивание с авторизацией через Bearer token

API часто используют токены в заголовке Authorization. Пример получения файла с GitHub Releases.

import requests

token = "ghp_xxxxxxxxxxxx"
headers = {"Authorization": f"Bearer {token}"}

url = "https://api.github.com/repos/owner/repo/releases/assets/12345"
response = requests.get(url, headers=headers, stream=True)
response.raise_for_status()

# Имя файла может быть из заголовка Content-Disposition
import re
cd = response.headers.get("Content-Disposition", "")
filename = re.findall('filename="(.+)"', cd)[0] if cd else "downloaded_file"

with open(filename, "wb") as f:
    for chunk in response.iter_content(8192):
        if chunk:
            f.write(chunk)

# Результат: файл (например, release.zip) сохранён с оригинальным именем

Извлечение имени из Content-Disposition позволяет сохранить файл с правильным расширением.

Скачивание с проверкой целостности через хэш

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

import requests
import hashlib

url = "https://example.com/debian.iso"
expected_hash = "a1b2c3d4e5f6..."  # SHA256 из официального сайта

response = requests.get(url, stream=True)
response.raise_for_status()

sha256 = hashlib.sha256()
with open("debian.iso", "wb") as f:
    for chunk in response.iter_content(8192):
        if chunk:
            f.write(chunk)
            sha256.update(chunk)

computed_hash = sha256.hexdigest()
if computed_hash == expected_hash:
    print("Хэш совпадает, файл корректен.")
else:
    print("Ошибка: хэш не совпадает, файл повреждён.")

# Результат: вывод сообщения о совпадении или несовпадении хэша

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

Параллельная загрузка нескольких файлов с помощью ThreadPoolExecutor

При необходимости скачать много файлов одновременно используется пул потоков из стандартной библиотеки concurrent.futures.

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

def download_file(url, dest_folder):
    local_filename = os.path.join(dest_folder, url.split("/")[-1])
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_filename, "wb") as f:
            for chunk in r.iter_content(8192):
                if chunk:
                    f.write(chunk)
    return local_filename

urls = [
    "https://example.com/photo1.jpg",
    "https://example.com/photo2.jpg",
    "https://example.com/photo3.jpg",
]

with ThreadPoolExecutor(max_workers=4) as executor:
    future_to_url = {executor.submit(download_file, url, "./downloads"): url for url in urls}
    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            result = future.result()
            print(f"Скачан: {result}")
        except Exception as e:
            print(f"Ошибка для {url}: {e}")

# Вывод:
# Скачан: ./downloads/photo1.jpg
# Скачан: ./downloads/photo2.jpg
# Скачан: ./downloads/photo3.jpg

Число потоков можно регулировать. Важно: параллельные запросы к одному серверу могут быть ограничены (rate limiting).

Скачивание с проверкой ETag для предотвращения повторной загрузки одинаковых файлов

Заголовок ETag представляет версию файла. Сохранив его в локальном кэше, можно в следующий раз отправить заголовок If-None-Match и получить ответ 304 Not Modified, если файл не изменился.

import requests
import json

url = "https://example.com/versioned_data.json"
cache_file = "etag_cache.json"

# Читаем сохранённый ETag
etag = None
try:
    with open(cache_file, "r") as cf:
        cache = json.load(cf)
        etag = cache.get(url)
except FileNotFoundError:
    pass

headers = {}
if etag:
    headers["If-None-Match"] = etag

response = requests.get(url, headers=headers)
if response.status_code == 304:
    print("Файл не изменился, используем локальную копию.")
else:
    with open("versioned_data.json", "wb") as f:
        f.write(response.content)
    # Сохраняем новый ETag
    new_etag = response.headers.get("ETag")
    if new_etag:
        with open(cache_file, "w") as cf:
            json.dump({url: new_etag}, cf)
        print("Файл обновлён.")

# Результат: либо сообщение "Файл не изменился", либо файл перезаписан

Аналогично можно использовать заголовок If-Modified-Since с датой Last-Modified.