Downloading Files with Progress in Python CLIs

Your CLI fetches something big — a dataset, a model file, a release artefact, a database snapshot. The naive version, open(dest, "wb").write(httpx.get(url).content), loads the entire file into memory, shows nothing for two minutes, and if the connection drops at 95% leaves a truncated file at the destination that looks perfectly valid until something tries to read it. The user's only recourse is to start again from zero. This guide builds a download function that streams in chunks, shows a Rich progress bar, writes to a temporary .part file, verifies a checksum, renames into place only when everything checks out, and resumes interrupted downloads with HTTP range requests. It is part of the HTTP APIs topic.

Prerequisites

• Python 3.10+, httpx and rich (uv add httpx rich).
• A server that reports Content-Length; resuming also needs Accept-Ranges: bytes. Most static file hosts, object stores and CDNs provide both.
• Ideally, an expected SHA-256 for the file — published alongside releases, or returned by your API.

The shape of a safe download

A download that cannot corrupt anything follows the same shape as an atomic write: the destination path is only ever touched by a final rename. Everything before that happens in a .part file beside it.

Streaming matters for memory: client.stream("GET", url) returns a response whose body you iterate in chunks, so a 20 GB file costs the same memory as a 20 KB one. Writing to .part matters for correctness: an interrupted download leaves a clearly unfinished file that the next run can resume or discard, never a truncated file with the real name.

The recipe

# src/mytool/download.py
from __future__ import annotations

import hashlib
import os
from pathlib import Path

import httpx
from rich.progress import (BarColumn, DownloadColumn, Progress, TextColumn,
                           TimeRemainingColumn, TransferSpeedColumn)

CHUNK = 64 * 1024


class DownloadError(Exception):
    pass


def _progress(enabled: bool) -> Progress:
    return Progress(
        TextColumn("[bold]{task.description}"), BarColumn(), DownloadColumn(),
        TransferSpeedColumn(), TimeRemainingColumn(),
        disable=not enabled, transient=True,
    )


def download(client: httpx.Client, url: str, dest: Path, *, sha256: str | None = None,
             show_progress: bool = True) -> Path:
    dest = dest.resolve()
    part = dest.with_name(dest.name + ".part")
    digest = hashlib.sha256()
    headers: dict[str, str] = {}
    offset = part.stat().st_size if part.exists() else 0
    if offset:
        headers["Range"] = f"bytes={offset}-"

    with client.stream("GET", url, headers=headers) as response:
        if response.status_code == 416:            # .part already complete or stale
            part.unlink()
            return download(client, url, dest, sha256=sha256, show_progress=show_progress)
        response.raise_for_status()
        resumed = response.status_code == 206
        if not resumed:
            offset = 0                               # server ignored Range: start over
        mode = "ab" if resumed else "wb"
        if resumed:                                   # hash the bytes we already have
            with part.open("rb") as existing:
                for block in iter(lambda: existing.read(CHUNK), b""):
                    digest.update(block)
        length = response.headers.get("Content-Length")
        total = offset + int(length) if length else None

        with _progress(show_progress) as progress, part.open(mode) as fh:
            task = progress.add_task(dest.name, total=total, completed=offset)
            for chunk in response.iter_bytes(CHUNK):
                fh.write(chunk)
                digest.update(chunk)
                progress.update(task, advance=len(chunk))
            fh.flush()
            os.fsync(fh.fileno())

    size = part.stat().st_size
    if total is not None and size != total:
        raise DownloadError(f"incomplete download: {size} of {total} bytes (run again to resume)")
    if sha256 and digest.hexdigest() != sha256.lower():
        part.unlink()
        raise DownloadError("checksum mismatch — the file was corrupted or changed; removed it")
    os.replace(part, dest)
    return dest

Some details that are easy to get wrong:

• iter_bytes, not iter_raw. iter_bytes() decodes any Content-Encoding (gzip) the server applied for transport, so you get the actual file. The trade-off is that Content-Length then describes the compressed size; static binaries are almost never transfer-compressed, so this rarely matters, but it is why the progress total can be approximate.
• A 206 Partial Content is the only proof a resume worked. Servers that do not support ranges simply ignore the header and send the whole file with 200. The code checks the status and starts over when that happens, instead of appending a complete file to a partial one.
• 416 Range Not Satisfiable means the .part file is at least as big as the resource — typically a finished download whose rename never happened, or a file that shrank on the server. Deleting it and starting again is the safe response.
• The checksum covers every byte, including those downloaded by an earlier run, which is why the existing .part content is hashed before appending.

For stronger guarantees on resume, send If-Range with the ETag from the original response. If the file changed on the server, the server then returns the whole new file (200) instead of splicing a range of the new version onto the old one. Store the ETag next to the .part file (for example in .part.etag) to make that work across runs.

The command

# src/mytool/cli.py
import sys
from pathlib import Path

import httpx
import typer

from mytool.download import DownloadError, download

app = typer.Typer()


@app.callback()
def main() -> None:
    """Dataset tools."""


@app.command()
def fetch(
    url: str,
    out: Path = typer.Option(None, "--out", "-o", help="Destination (default: name from URL)."),
    sha256: str = typer.Option(None, help="Expected SHA-256 of the file."),
) -> None:
    """Download URL, resuming if a previous attempt was interrupted."""
    dest = out or Path(httpx.URL(url).path.rsplit("/", 1)[-1] or "download")
    timeout = httpx.Timeout(60.0, connect=10.0)
    with httpx.Client(timeout=timeout, follow_redirects=True) as client:
        try:
            path = download(client, url, dest, sha256=sha256, show_progress=sys.stderr.isatty())
        except (DownloadError, httpx.HTTPError) as exc:
            typer.secho(f"error: {exc}", fg="red", err=True)
            raise typer.Exit(1)
        except KeyboardInterrupt:
            typer.echo("\ninterrupted — run the same command again to resume", err=True)
            raise typer.Exit(130)
    if sha256:
        typer.echo(f"verified sha256 {sha256[:4]}...{sha256[-4:]}", err=True)
    typer.echo(f"saved {path}", err=True)


if __name__ == "__main__":
    app()

follow_redirects=True is important for downloads: release assets on GitHub, object-store presigned URLs and CDNs almost always redirect. The read timeout of 60 seconds limits the gap between chunks, not the total download time, so a large file on a slow link is fine as long as bytes keep arriving.

UX considerations

• Progress only on a terminal. Rich's disable= keeps the bar off when stderr is redirected, so CI logs do not fill with carriage-return redraws. transient=True removes the bar when done, leaving a clean "saved" line. The general rule is in detecting a TTY and adapting output.
• Show size, speed and time remaining. Those three columns answer "is it working?" and "should I get coffee?" — which is all a progress bar is for. More patterns are in adding progress bars and spinners to Python CLIs.
• Tell the user resuming is possible. On Ctrl+C or a dropped connection, "run the same command again to resume" converts a frustrating failure into a minor pause.
• Verify, and say so. Printing a short form of the verified hash reassures users that the file is exactly what was published.
• Refuse to overwrite silently. If the destination exists and differs, consider requiring --force, or at least say that it was replaced.

Testing the behaviour

A mock transport that serves a known byte string and honours Range lets you test full downloads, resumes and corruption without a network:

# tests/test_download.py
import hashlib

import httpx
import pytest

from mytool.download import DownloadError, download

BLOB = bytes(range(256)) * 1000          # 256 KB of known data
SHA = hashlib.sha256(BLOB).hexdigest()


def server(ranges: bool = True, body: bytes = BLOB):
    def handler(request):
        rng = request.headers.get("Range")
        if ranges and rng:
            start = int(rng.removeprefix("bytes=").rstrip("-"))
            if start >= len(body):
                return httpx.Response(416)
            chunk = body[start:]
            return httpx.Response(206, content=chunk, headers={"Content-Length": str(len(chunk))})
        return httpx.Response(200, content=body, headers={"Content-Length": str(len(body))})
    return httpx.Client(transport=httpx.MockTransport(handler))


def test_full_download(tmp_path):
    dest = download(server(), "https://files.test/data.bin", tmp_path / "data.bin",
                    sha256=SHA, show_progress=False)
    assert dest.read_bytes() == BLOB
    assert not (tmp_path / "data.bin.part").exists()


def test_resume_appends_the_rest(tmp_path):
    (tmp_path / "data.bin.part").write_bytes(BLOB[:100_000])
    dest = download(server(), "https://files.test/data.bin", tmp_path / "data.bin",
                    sha256=SHA, show_progress=False)
    assert dest.read_bytes() == BLOB


def test_server_without_ranges_restarts(tmp_path):
    (tmp_path / "data.bin.part").write_bytes(b"garbage" * 10)
    dest = download(server(ranges=False), "https://files.test/data.bin", tmp_path / "data.bin",
                    sha256=SHA, show_progress=False)
    assert dest.read_bytes() == BLOB


def test_bad_checksum_leaves_no_file(tmp_path):
    with pytest.raises(DownloadError, match="checksum"):
        download(server(), "https://files.test/data.bin", tmp_path / "data.bin",
                 sha256="0" * 64, show_progress=False)
    assert list(tmp_path.iterdir()) == []

The "without ranges" test is the one that catches the classic bug of appending a full 200 response to a stale partial file. For wider coverage of network edge cases, the scripted-transport pattern from retries and backoff for CLI HTTP calls works here too.

Conclusion

A dependable download streams in chunks, reports progress only where someone is watching, writes to a .part file, verifies size and checksum, and renames into place as its last act. Resume by sending Range, trust only a 206, and recover from 416. Each piece is a few lines; together they turn "the download failed at 95%" from a lost afternoon into a single rerun.

Frequently asked questions

Related

• Up: Calling HTTP APIs from Python CLIs
• Retries and backoff for CLI HTTP calls
• Writing files atomically in Python CLIs
• Adding progress bars and spinners to Python CLIs
• Building an API client CLI with httpx