Uploading Files with Multipart and Progress in a Python CLI

Uploads are the mirror image of downloads, and CLIs need them as often: publishing a build artifact, attaching a log to a support ticket, pushing a dataset, sending a report to an API. Two request shapes cover almost every case. Multipart form data — multipart/form-data, what an HTML file input sends — carries one or more files plus ordinary form fields in a single POST; most application APIs use it. A raw body — the file’s bytes as the entire request, usually a PUT — is what object stores and presigned URLs (S3, GCS, Azure Blob) expect. Both should stream from disk rather than load the file into memory, report progress for anything larger than a few megabytes, and fail with a message that says what happened. This guide implements both with httpx, adds a byte-accurate progress bar, and tests the requests without a server. It belongs to the calling HTTP APIs from Python CLIs topic.

Prerequisites

• uv add httpx rich typer, plus respx for tests (examples checked with httpx 0.28 and respx 0.23).
• The client conventions from building an API client CLI with httpx.

Two shapes of upload

A multipart body is a sequence of parts separated by a random boundary string; each part has headers — Content-Disposition: form-data; name="file"; filename="report.pdf" and a Content-Type — followed by its content. The server reads named fields and files from it, which is why APIs that need metadata alongside the file (a project name, a description) use it. A raw upload has no structure at all: the body is the file, the Content-Type header describes it, and any metadata travels in the URL or other headers. Presigned URLs are raw uploads by design — the signature covers a specific method, path and often the content type, so send exactly what the URL was signed for.

The recipe

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

import mimetypes
import os
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import BinaryIO

import httpx

CHUNK = 64 * 1024


class ProgressFile:
    """File wrapper that reports bytes as httpx reads them (for multipart uploads)."""

    def __init__(self, f: BinaryIO, on_bytes: Callable[[int], None]) -> None:
        self._f, self._on_bytes = f, on_bytes

    def read(self, size: int = -1) -> bytes:
        data = self._f.read(size)
        self._on_bytes(len(data))
        return data

    def fileno(self) -> int:                 # lets httpx compute Content-Length via fstat
        return self._f.fileno()

    def seek(self, offset: int, whence: int = os.SEEK_SET) -> int:
        return self._f.seek(offset, whence)

    def tell(self) -> int:
        return self._f.tell()


def upload_multipart(client: httpx.Client, url: str, path: Path, *, fields: dict[str, str],
                     on_bytes: Callable[[int], None]) -> httpx.Response:
    """POST PATH as multipart/form-data with extra form FIELDS, streaming from disk."""
    content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    with path.open("rb") as f:
        files = {"file": (path.name, ProgressFile(f, on_bytes), content_type)}
        response = client.post(url, data=fields, files=files)
    response.raise_for_status()
    return response


def _chunks(path: Path, on_bytes: Callable[[int], None]) -> Iterator[bytes]:
    with path.open("rb") as f:
        while chunk := f.read(CHUNK):
            on_bytes(len(chunk))
            yield chunk


def upload_raw(client: httpx.Client, url: str, path: Path, *,
               on_bytes: Callable[[int], None]) -> httpx.Response:
    """PUT PATH as the raw request body (presigned URLs, object stores)."""
    headers = {"Content-Length": str(path.stat().st_size),        # avoid chunked encoding
               "Content-Type": mimetypes.guess_type(path.name)[0] or "application/octet-stream"}
    response = client.put(url, content=_chunks(path, on_bytes), headers=headers)
    response.raise_for_status()
    return response

Streaming multipart with progress

httpx builds multipart bodies lazily: when a file part is an open file object, it reads it in chunks while sending, so memory use stays flat for any file size. To observe that reading, ProgressFile wraps the real file and calls on_bytes with the size of every chunk httpx reads. It also exposes fileno, seek and tell, which httpx uses to determine the file’s length up front: with a known length the request gets a Content-Length header, which some servers require, instead of chunked transfer encoding. The content type comes from the file extension via mimetypes, falling back to application/octet-stream.

Progress measured this way is “bytes handed to the network layer”, which runs slightly ahead of what the server has received — operating system buffers hold a little data in flight. For a progress bar that is accurate enough; for the final answer, rely on the response status.

Raw uploads with a generator

upload_raw passes a generator as content=. httpx sends each yielded chunk as it is produced, so the generator is a natural place to count bytes. A generator has no length, so httpx would normally fall back to chunked encoding — which many object stores reject for PUT. Setting Content-Length explicitly from the file size avoids that. Do not change the file while uploading: if the size no longer matches, the request fails or the stored object is truncated.

The command

# src/mytool/cli.py
from pathlib import Path
from typing import Annotated

import httpx
import typer
from rich.console import Console
from rich.progress import BarColumn, DownloadColumn, Progress, TransferSpeedColumn

from mytool.upload import upload_multipart

app = typer.Typer()
API = "https://api.example.com"


@app.callback()
def main() -> None:
    """Artifact tool."""


@app.command()
def upload(path: Annotated[Path, typer.Argument(exists=True, dir_okay=False)],
           project: str = "default") -> None:
    """Upload PATH as an artifact."""
    err = Console(stderr=True)
    columns = ("{task.description}", BarColumn(), DownloadColumn(), TransferSpeedColumn())
    timeout = httpx.Timeout(10.0, write=60.0)             # per chunk, not for the whole file
    with httpx.Client(base_url=API, timeout=timeout) as client, \
            Progress(*columns, console=err, transient=True) as progress:
        task = progress.add_task(path.name, total=path.stat().st_size)
        try:
            response = upload_multipart(client, "/artifacts", path, fields={"project": project},
                                        on_bytes=lambda n: progress.advance(task, n))
        except httpx.HTTPStatusError as exc:
            err.print(f"error: upload rejected ({exc.response.status_code}): {exc.response.text[:200]}")
            raise typer.Exit(1)
        except httpx.TransportError as exc:
            err.print(f"error: upload failed: {exc}")
            raise typer.Exit(1)
    typer.echo(response.json()["id"])

The progress bar goes to stderr and is transient, so on success only the new artifact’s ID remains on stdout — ready for ID=$(mytool upload report.pdf). The timeout is set per operation: ten seconds to connect and read the response, sixty for each write, which bounds a stalled connection without imposing an overall limit that a large file on a slow line would exceed. Two error branches cover what users meet: the server rejecting the upload (size limits, permissions, validation) shows the status and the start of the response body; a transport error (network drop, timeout, TLS) says the upload failed.

Large files and retries

For files of hundreds of megabytes or more, a single request is fragile: a dropped connection at 95% means starting again. Object stores offer multipart uploads for this (confusingly, unrelated to multipart form data): the file is split into parts of, say, 16 MB, each uploaded with its own request and retried independently, then the parts are combined with a final call. Other APIs offer resumable upload protocols such as tus. If your API supports either, use it above a size threshold. Retrying a whole single-request upload is reasonable for small files, but only for idempotent requests — a PUT to a fixed URL is safe to repeat, a POST that creates a new artifact each time may produce duplicates. The retry rules in retries and backoff for CLI HTTP calls apply, and a retry must reopen or rewind the file.

UX considerations

• Check before sending. If the API documents a size limit, compare against path.stat().st_size first and fail immediately instead of after uploading a gigabyte.
• Show speed and size. DownloadColumn and TransferSpeedColumn work for uploads too; “12.4/80.0 MB, 3.1 MB/s” tells users whether to wait or get coffee.
• Print the result, not the progress, on stdout. The ID or URL of the uploaded object is the command’s output.
• Keep progress off in pipes and CI — Rich does this automatically when stderr is not a terminal.
• Never log presigned URLs. Their query string is a credential until it expires; redact it as in redacting secrets from CLI output and logs.

Testing the behaviour

httpx.MockTransport hands each request to a function, which can inspect the exact bytes that would have been sent — the boundary, the part headers, the form field, the length headers — and the progress callback is checked to account for every byte. respx covers the command end to end:

# tests/test_upload.py
import httpx
import respx
from typer.testing import CliRunner

from mytool.cli import app
from mytool.upload import upload_multipart, upload_raw


def make_file(tmp_path, size=300_000):
    path = tmp_path / "report.pdf"
    path.write_bytes(bytes(range(256)) * (size // 256) + b"x" * (size % 256))
    return path


def test_multipart_body_fields_and_progress(tmp_path):
    path = make_file(tmp_path)
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        body = request.read()
        seen["type"] = request.headers["content-type"]
        seen["length"] = int(request.headers["content-length"])
        seen["has_file"] = b'filename="report.pdf"' in body and b"application/pdf" in body
        seen["has_field"] = b'name="project"' in body and b"docs" in body
        return httpx.Response(201, json={"id": "art_1"})

    reported = []
    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
        upload_multipart(client, "https://x/upload", path, fields={"project": "docs"},
                         on_bytes=reported.append)
    assert seen["type"].startswith("multipart/form-data; boundary=")
    assert seen["length"] > path.stat().st_size and seen["has_file"] and seen["has_field"]
    assert sum(reported) == path.stat().st_size


def test_raw_put_streams_with_content_length(tmp_path):
    path = make_file(tmp_path)
    seen = {}

    def handler(request):
        seen["length"] = request.headers.get("content-length")
        seen["chunked"] = request.headers.get("transfer-encoding")
        seen["body"] = request.read()
        return httpx.Response(200)

    reported = []
    with httpx.Client(transport=httpx.MockTransport(handler)) as client:
        upload_raw(client, "https://bucket/obj?sig=1", path, on_bytes=reported.append)
    assert seen["length"] == str(path.stat().st_size) and seen["chunked"] is None
    assert seen["body"] == path.read_bytes()
    assert len(reported) == 5                                   # 64 KiB chunks


@respx.mock
def test_cli_prints_id_and_reports_rejection(tmp_path):
    path = make_file(tmp_path, 1000)
    route = respx.post("https://api.example.com/artifacts")
    route.return_value = httpx.Response(201, json={"id": "art_42"})
    ok = CliRunner().invoke(app, ["upload", str(path)])
    assert ok.exit_code == 0 and ok.stdout == "art_42\n"

    route.return_value = httpx.Response(413, text="file too large")
    bad = CliRunner().invoke(app, ["upload", str(path)])
    assert bad.exit_code == 1 and "upload rejected (413): file too large" in bad.stderr

The raw-upload test asserts that Content-Length is set and Transfer-Encoding is absent, which is the property object stores care about, and that the file was read in five 64 KiB chunks — proof that it streamed rather than being read whole. More on mocking in mocking HTTP in CLI tests with respx.

Conclusion

Use multipart form data when an API wants a file plus fields, and a raw PUT with an explicit Content-Length for object stores and presigned URLs. Stream from disk in both cases — a wrapper file object for multipart, a generator for raw bodies — and count bytes in the same place to drive a transient Rich progress bar on stderr. Set per-operation timeouts, report server rejections and transport failures distinctly, print only the result on stdout, switch to chunked or resumable protocols for very large files, and test the exact request bytes with MockTransport.

Frequently asked questions

Related

• Up: Calling HTTP APIs from Python CLIs
• Downloading files with progress in Python
• Retries and backoff for CLI HTTP calls
• Mocking HTTP in CLI tests with respx
• Showing progress for concurrent tasks