Starlette multipart form uploads parsed with python-multipart, tested offline: working versions (tested 2026-10-15)

python-multipart==0.0.32
starlette==1.8.0
httpx==0.28.1
pytest==9.1.1
pytest-asyncio==1.4.0

from __future__ import annotations

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route


async def inspect_upload(request: Request) -> JSONResponse:
    # request.form() is an async context manager, so the UploadFile handles are closed on exit.
    async with request.form() as form:
        files = form.getlist("files")
        contents = [await item.read() for item in files]
        return JSONResponse(
            {
                "first": files[0].filename if files else None,
                "names": [item.filename for item in files],
                "types": [item.content_type for item in files],
                "sizes": [item.size for item in files],
                "texts": [data.decode() for data in contents],
                "tags": form.getlist("tags"),
                "note": form.get("note"),
            }
        )


async def digest(request: Request) -> JSONResponse:
    form = await request.form()
    item = form["files"]
    # read() advances the underlying SpooledTemporaryFile, so read again only after seek(0).
    head = await item.read(4)
    rest = await item.read()
    await item.seek(0)
    whole = await item.read()
    return JSONResponse({"head": head.decode(), "rest": rest.decode(), "whole": whole.decode()})


async def capped(request: Request) -> JSONResponse:
    # A MultiPartException raised inside a routed request turns into an HTTP 400.
    form = await request.form(max_files=1)
    return JSONResponse({"count": len(form.getlist("files"))})


app = Starlette(
    routes=[
        Route("/inspect", inspect_upload, methods=["POST"]),
        Route("/digest", digest, methods=["POST"]),
        Route("/capped", capped, methods=["POST"]),
    ]
)

import httpx
import pytest

from app import app


@pytest.fixture
async def client():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as c:
        yield c


async def test_single_upload_is_parsed(client):
    r = await client.post(
        "/inspect",
        data={"note": "hi"},
        files={"files": ("a.txt", b"hello", "text/plain")},
    )
    assert r.status_code == 200
    body = r.json()
    assert body["first"] == "a.txt"
    assert body["names"] == ["a.txt"]
    assert body["types"] == ["text/plain"]
    assert body["sizes"] == [5]
    assert body["texts"] == ["hello"]
    assert body["note"] == "hi"


async def test_repeated_file_field_first_vs_getlist(client):
    files = [
        ("files", ("a.txt", b"1", "text/plain")),
        ("files", ("b.txt", b"22", "text/plain")),
    ]
    r = await client.post("/inspect", files=files)
    body = r.json()
    # form["files"] keeps only the first part; getlist returns every part.
    assert body["first"] == "a.txt"
    assert body["names"] == ["a.txt", "b.txt"]
    assert body["sizes"] == [1, 2]


async def test_repeated_text_field_getlist(client):
    r = await client.post("/inspect", data={"tags": ["x", "y"]})
    body = r.json()
    assert body["tags"] == ["x", "y"]
    assert body["names"] == []
    assert body["first"] is None


async def test_reread_requires_seek(client):
    r = await client.post("/digest", files={"files": ("a.txt", b"abcdef", "text/plain")})
    assert r.json() == {"head": "abcd", "rest": "ef", "whole": "abcdef"}


async def test_max_files_limit_becomes_400(client):
    files = [
        ("files", ("a.txt", b"1", "text/plain")),
        ("files", ("b.txt", b"2", "text/plain")),
    ]
    r = await client.post("/capped", files=files)
    assert r.status_code == 400
    # Plain Starlette renders HTTPException as text, not JSON like FastAPI does.
    assert r.text == "Too many files. Maximum number of files is 1."

[pytest]
pythonpath = .
asyncio_mode = auto