Загрузка файла оборвалась: поток HTTPX, временный файл и закрытие ответа

Загрузчик получает большой файл и сразу пишет response.content в конечный путь. В этом варианте весь ответ сначала оказывается в памяти. Если перейти к записи кусками прямо в конечный файл, появится другая проблема: после обрыва сети потребитель увидит недокачанный результат как готовый. Нужно одновременно определить чтение потока, закрытие ответа и момент публикации файла.

В HTTPX потоковый ответ удобно открывать через async with client.stream. Внутри сначала проверяем статус через raise_for_status, затем читаем aiter_bytes. Контекст закроет ответ и при исключении. На диск пишем уникальный временный файл рядом с назначением, а конечное имя заменяем лишь после успешного завершения чтения.

Чтение до публикации файла

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

import asyncio
import os
import tempfile
from pathlib import Path
import httpx

async def download(client, url: str, destination: Path):
    temporary = None
    try:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            with tempfile.NamedTemporaryFile(dir=destination.parent, delete=False) as stream:
                temporary = Path(stream.name)
                async for chunk in response.aiter_bytes():
                    stream.write(chunk)
        os.replace(temporary, destination)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)

async def main():
    transport = httpx.MockTransport(lambda request: httpx.Response(200, content=b"new"))
    with tempfile.TemporaryDirectory() as directory:
        destination = Path(directory) / "download.bin"
        destination.write_bytes(b"old")
        async with httpx.AsyncClient(transport=transport) as client:
            await download(client, "https://files.invalid/data", destination)
        assert destination.read_bytes() == b"new"
        print("downloaded")

asyncio.run(main())

Обрыв, размер и время загрузки

Установите HTTPX: python -m pip install httpx. На Python 3.12 пример печатает downloaded. Для отрицательного теста передайте поток, который после первой порции поднимает httpx.ReadError. Старый конечный файл должен сохраниться, временный — исчезнуть. Отдельно проверьте 404: содержимое страницы ошибки не должно стать загруженным документом.

Обычная запись в файл синхронна. В примере она короткая, но на медленном диске или при больших блоках может задерживать event loop. Для такого окружения вынесите дисковые операции в подходящий исполнитель или используйте файловый API, принятый в проекте. При этом сохраните порядок: запись всех порций, закрытие файла, публикация, уборка при отказе.

aiter_bytes возвращает декодированное содержимое после обработки HTTP content encoding. Если задача состоит в сохранении исходного сжатого потока, разберите aiter_raw и ожидаемый формат. Имя метода здесь влияет на то, какие именно байты попадут в файл. Для проверки целостности используйте контрольную сумму или размер согласно договорённости с источником, учитывая выбранное представление.

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

Когда локальный пример работает, полезно разобрать соседние механизмы языка и их взаимодействие. В курсе «Python для профессионалов» есть практика по итераторам, генераторам, декораторам, asyncio, метапрограммированию и профилированию. Эти темы помогают понимать время жизни объектов, управление выполнением и причины поведения прикладной программы.

Посмотреть программу и практику курса.

Документация: официальный источник по теме.

Другие задачи по теме: База, HTTP-клиенты и тесты API.