Как сохранить полученный от сервера ответ в файл

Дополнительные примеры сохранения ответа в файл

Сохранение с проверкой целостности (хэш SHA256)

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

import requests
import hashlib

url = 'https://example.com/firmware.bin'
expected_hash = 'a1b2c3...'  # заранее известный хэш

response = requests.get(url, stream=True)
sha256 = hashlib.sha256()

with open('firmware.bin', 'wb') as f:
    for chunk in response.iter_content(chunk_size=4096):
        f.write(chunk)
        sha256.update(chunk)

actual_hash = sha256.hexdigest()
if actual_hash == expected_hash:
    print('Целостность подтверждена')
else:
    print('Хэш не совпадает - файл повреждён')

Целостность подтверждена

Такой подход применяется для загрузки дистрибутивов, прошивок и других критичных данных.

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

Для ускорения можно загружать несколько файлов одновременно:

import requests
from concurrent.futures import ThreadPoolExecutor

def download_file(url, filename):
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    return f'{filename} загружен'

urls = [
    ('https://httpbin.org/image/png', '1.png'),
    ('https://httpbin.org/image/jpeg', '1.jpg'),
    ('https://httpbin.org/image/svg', '1.svg')
]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(lambda urls: download_file(*urls), urls))

for res in results:
    print(res)

1.png загружен
1.jpg загружен
1.svg загружен

Ограничение: не следует открывать слишком много потоков, чтобы не перегрузить сетевое соединение.

Сохранение ответа с автоповтором при ошибках (retry)

Используя requests.adapters.HTTPAdapter с политикой повторных попыток:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retry_policy = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry_policy)
session.mount('https://', adapter)
session.mount('http://', adapter)

url = 'https://example.com/unstable-server/document.pdf'
response = session.get(url, stream=True)
response.raise_for_status()

with open('document.pdf', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        f.write(chunk)
print('Документ сохранён после возможных повторов')

Повторные попытки увеличивают надёжность при нестабильной сети.

Сохранение ответа с использованием shutil.copyfileobj

Библиотека shutil предоставляет удобный способ копирования потока:

import requests
import shutil

url = 'https://httpbin.org/image/png'
response = requests.get(url, stream=True)
response.raw.decode_content = True  # обрабатывает сжатие gzip

with open('image_shutil.png', 'wb') as f:
    shutil.copyfileobj(response.raw, f)
print('Файл сохранён через shutil')

Этот метод компактен, но требует осторожности с кодировкой (response.raw может содержать несжатые данные).

Сохранение ответа в файл в temp-директорию

Для временных файлов используют tempfile:

import requests
import tempfile
import os

url = 'https://httpbin.org/image/png'
response = requests.get(url, stream=True)

with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as tmp:
    for chunk in response.iter_content(chunk_size=8192):
        tmp.write(chunk)
    tmp_path = tmp.name
    print(f'Временный файл: {tmp_path}')

# Позже можно удалить вручную
os.unlink(tmp_path)

Удобно для кэширования или промежуточной обработки.