Reliable File Downloads in Python: Streaming, Size Limits, and SHA-256

An automation script opens data.csv for writing before fetching it. When the connection breaks, the previous complete file is gone and downstream steps may consume a partial result. This tutorial uses Python’s standard library to stream a download, enforce a size limit, verify SHA-256, and replace the destination only after every check passes.

It is intended for beginner-to-intermediate readers downloading known data exports or release files. It makes one attempt, with no resume or automatic retries. For retrying a small JSON API request, see HTTP JSON timeouts and bounded retries.

1. Method: validate before replacement

The download follows these steps:

1. Accept only HTTP(S) URLs and HTTP 200. Reject redirects to keep the source explicit.
2. Create a random sibling .part file and read at most 64 KiB per chunk.
3. Count bytes and update SHA-256; also check Content-Length when present.
4. After size, length, and trusted checksum checks pass, flush and close the temporary file, then replace the destination.
5. Clean up the temporary file on catchable failures and return a nonzero exit code.

Content-Length can be absent, and length alone cannot establish correctness. Always enforce the byte limit and SHA-256 check. Obtain the expected checksum through a trusted publisher channel; hashing the downloaded result and comparing it to itself proves nothing. For existing file inventories, see SHA-256 dataset manifests.

2. Setup

Use Python 3.10 or newer. The commands that fetch the example scripts also need curl. No third-party Python packages are required. The code and failure tests were verified on Linux with Python 3.11.4. Start in a new practice directory:

python3 --version
mkdir stream-download-demo
cd stream-download-demo
curl -fS https://www.bobobk.com/downloads/stream-download/download_file.py -o download_file.py
curl -fS https://www.bobobk.com/downloads/stream-download/selftest.py -o selftest.py
python3 download_file.py --help

Alternatively, save the complete code below as download_file.py. The accompanying selftest.py is a complete local test and does not fetch external data.

The destination’s parent directory must already exist. Use a trusted directory you control and a single writer per destination.

3. Complete implementation

"""Download one known HTTP(S) file and verify SHA-256 before replacing it."""
import argparse
import hashlib
from http.client import HTTPException
import json
import math
import os
from pathlib import Path
import re
import sys
import tempfile
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def download(url, target, expected_sha256, max_bytes, timeout):
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"} or not parts.hostname:
        raise ValueError("URL must use http:// or https:// with a host")
    target = Path(target).absolute()
    temporary = None
    request = Request(url, headers={"Accept-Encoding": "identity"})
    try:
        with build_opener(NoRedirect()).open(request, timeout=timeout) as response:
            if response.status != 200:
                raise ValueError(f"expected HTTP 200, got {response.status}")
            if response.headers.get("Content-Encoding", "identity").lower() != "identity":
                raise ValueError("unexpected Content-Encoding")
            lengths = response.headers.get_all("Content-Length", [])
            if lengths and (len(lengths) != 1 or not re.fullmatch(r"[0-9]+", lengths[0])):
                raise ValueError("invalid Content-Length")
            declared = int(lengths[0]) if lengths else None
            if declared is not None and declared > max_bytes:
                raise ValueError("Content-Length exceeds --max-bytes")
            digest = hashlib.sha256()
            total = 0
            with tempfile.NamedTemporaryFile(
                dir=target.parent, prefix=f".{target.name}.",
                suffix=".part", delete=False,
            ) as output:
                temporary = Path(output.name)
                while True:
                    chunk = response.read(min(65536, max_bytes - total + 1))
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > max_bytes:
                        raise ValueError("body exceeds --max-bytes")
                    output.write(chunk)
                    digest.update(chunk)
                if declared is not None and total != declared:
                    raise ValueError("Content-Length mismatch (truncated response)")
                actual = digest.hexdigest()
                if actual != expected_sha256:
                    raise ValueError("SHA-256 mismatch")
                output.flush()
                os.fsync(output.fileno())
        os.replace(temporary, target)
        temporary = None
        return {"file": str(target), "bytes": total, "sha256": actual}
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("url")
    parser.add_argument("output", type=Path)
    parser.add_argument("--sha256", required=True, help="trusted 64-digit SHA-256")
    parser.add_argument("--max-bytes", type=int, required=True)
    parser.add_argument("--timeout", type=float, default=10.0,
                        help="blocking socket operation timeout, seconds (default: 10)")
    args = parser.parse_args()
    if not re.fullmatch(r"[0-9a-fA-F]{64}", args.sha256):
        parser.error("--sha256 must have exactly 64 hexadecimal digits")
    if args.max_bytes < 1:
        parser.error("--max-bytes must be positive")
    if not math.isfinite(args.timeout) or args.timeout <= 0:
        parser.error("--timeout must be finite and positive")
    try:
        result = download(args.url, args.output, args.sha256.lower(),
                          args.max_bytes, args.timeout)
    except (OSError, ValueError, HTTPException) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

NamedTemporaryFile creates the temporary file; dir=target.parent puts it beside the destination. A successful same-filesystem os.replace() replacement is atomic. The same saving pattern works for atomic JSON writes.

hashlib.update() updates the digest incrementally without loading the whole file. --max-bytes limits response-body bytes written; at the boundary, the script reads at most one additional byte to detect excess. It requests identity encoding and rejects a non-identity Content-Encoding, avoiding comparisons between an encoded response and a checksum for a different representation.

The urlopen timeout documentation describes the underlying blocking network operations also used by this opener. --timeout 3 is not a three-second deadline for the entire download: a slow transfer that keeps delivering data can take longer, and DNS resolution is not necessarily bounded by this setting. System proxy settings and default HTTPS certificate verification remain in use. Use only URLs you approve; this client does not accept untrusted user-supplied addresses.

4. Run and verify locally

1. In terminal A, create a test source and start a local server from the practice directory:

mkdir source
printf 'id,value\n1,42\n' > source/data.csv
python3 -m http.server 8765 --bind 127.0.0.1 --directory source

Leave terminal A running. The server listens only on localhost. If port 8765 is occupied, change it in both the server command and the URL below. Stop the server with Ctrl+C after the exercise.

2. In terminal B, enter the same stream-download-demo directory and run:

export no_proxy=127.0.0.1,localhost
export NO_PROXY=127.0.0.1,localhost
download_sha=$(python3 -c 'import hashlib; from pathlib import Path; print(hashlib.sha256(Path("source/data.csv").read_bytes()).hexdigest())')
python3 download_file.py http://127.0.0.1:8765/data.csv result.csv \
  --sha256 "$download_sha" --max-bytes 1048576 --timeout 3
python3 - <<'PY_CHECK'
from pathlib import Path
assert Path("result.csv").read_bytes() == b"id,value\n1,42\n"
assert not list(Path(".").glob(".result.csv.*.part"))
print("PASS: exact contents; no partial file")
PY_CHECK

The checksum comes from the source file you just created, demonstrating a trusted expected value. For real downloads, use the publisher’s checksum. The proxy environment variables affect this terminal and its children, bypassing proxies for the local test.

Success prints one JSON line with the absolute destination path, bytes: 14, and sha256, followed by PASS: exact contents; no partial file. Running it again replaces result.csv with the verified contents.

5. Failure tests: prove that old output survives

Run the companion test in terminal B. It starts its own local HTTP server on an available port, uses a temporary directory, and closes the server and deletes its test files afterward:

python3 selftest.py

Expected output:

PASS: 2 successes, 8 failures preserve old output; no partial files
PASS: failed first download leaves no output; 3 invalid arguments rejected

The tests cover successful downloads with and without Content-Length, plus a wrong hash, truncation, declared and actual size overflows, 404, 302, unexpected encoding, and timeout. Each failure checks the exit code, stderr, old file contents, and .part cleanup. Further checks verify that a failed first download creates no destination and invalid arguments return 2.

Controlled responses reproduce failures without disconnecting a real network. These tests leave the section 4 result.csv unchanged.

6. Automation and troubleshooting

In terminal B, reuse the section 4 variable and local server. Use && so the next step runs only after a successful download:

python3 download_file.py http://127.0.0.1:8765/data.csv result.csv \
  --sha256 "$download_sha" --max-bytes 1048576 --timeout 3 \
  && python3 -c 'from pathlib import Path; print(Path("result.csv").read_text())'

In production, supply your actual URL, destination, trusted checksum, and limit through the existing task entry point. A checksum failure must stop this run even when the previous file still exists; file existence is not download success.

Temporary files on Linux are normally readable and writable only by their owner; replacement does not inherit old file permissions. Explicitly set appropriate permissions within a trusted workflow if another process needs access. Atomic replacement lets readers see a complete old or new file; it does not establish business validity, order concurrent writers, or guarantee directory durability after power loss. finally cannot run after SIGKILL, a process crash, or power loss, so random .part files may remain. Confirm no download is active before inspecting and removing the matching leftovers.

7. Summary

Delay replacement until downloading and verification finish to preserve the last complete result during network failures. Chunked reads bound memory, the byte limit bounds writes, and a trusted SHA-256 checks content. Run the local failure tests, then require a successful exit before continuing automation.