Forms & File Uploads

uv add python-multipart
# or: pip install python-multipart

from typing import Annotated

from fastapi import FastAPI, Form

app = FastAPI(title="Bookshelf API")


@app.post("/reviews/form")
async def create_review(
    reviewer: Annotated[str, Form(min_length=2, max_length=40)],
    rating: Annotated[int, Form(ge=1, le=5)],
    book_id: Annotated[int, Form(alias="book-id")],
    tags: Annotated[list[str] | None, Form()] = None,
    text: Annotated[str | None, Form(max_length=2000)] = None,
) -> dict:
    return {"reviewer": reviewer, "rating": rating, "book_id": book_id, "tags": tags, "text": text}

curl -X POST localhost:8000/reviews/form \
  -d "reviewer=Ana" -d "rating=5" -d "book-id=7" -d "tags=scifi" -d "tags=classic"

{"reviewer": "Ana", "rating": 5, "book_id": 7, "tags": ["scifi", "classic"], "text": null}

{
  "detail": [
    {"type": "string_too_short", "loc": ["body", "reviewer"], "msg": "String should have at least 2 characters",
     "input": "A", "ctx": {"min_length": 2}},
    {"type": "less_than_equal", "loc": ["body", "rating"], "msg": "Input should be less than or equal to 5",
     "input": "9", "ctx": {"le": 5}},
    {"type": "missing", "loc": ["body", "book-id"], "msg": "Field required", "input": null}
  ]
}

from typing import Annotated, Literal

from fastapi import FastAPI, Form
from pydantic import BaseModel, Field

app = FastAPI(title="Bookshelf API")


class ReviewForm(BaseModel):
    model_config = {"extra": "forbid"}

    reviewer: Annotated[str, Field(min_length=2, max_length=40)]
    rating: Annotated[int, Field(ge=1, le=5)]
    spoiler: bool = False
    language: Literal["en", "es", "fr"] = "en"
    tags: list[str] = Field(default_factory=list)


@app.post("/reviews/form-model")
async def create_review(review: Annotated[ReviewForm, Form()]) -> ReviewForm:
    return review

curl -X POST localhost:8000/reviews/form-model -d "reviewer=Ana" -d "rating=5" -d "spoiler=true" -d "tags=scifi" -d "tags=classic"

{"reviewer": "Ana", "rating": 5, "spoiler": true, "language": "en", "tags": ["scifi", "classic"]}

{"detail": [{"type": "extra_forbidden", "loc": ["body", "extra"], "msg": "Extra inputs are not permitted", "input": "nope"}]}

from typing import Annotated

from fastapi import FastAPI, File, UploadFile

app = FastAPI(title="Bookshelf API")


@app.post("/covers/bytes")
async def cover_as_bytes(cover: Annotated[bytes, File(description="Cover image read fully into memory")]) -> dict:
    return {"size": len(cover)}


@app.post("/covers/upload")
async def cover_as_upload_file(
    cover: Annotated[UploadFile, File(description="Cover image as a spooled file")],
) -> dict:
    head = await cover.read(4)  # first 4 bytes
    await cover.seek(0)  # rewind
    whole = await cover.read()
    return {
        "filename": cover.filename,
        "content_type": cover.content_type,
        "size": cover.size,
        "headers": dict(cover.headers),
        "first_bytes": head.hex(),
        "bytes_read": len(whole),
    }


@app.post("/covers/sync")
def cover_in_a_sync_function(cover: UploadFile) -> dict:
    # In a plain `def` function the underlying SpooledTemporaryFile can be used directly.
    return {"first_line": cover.file.readline().decode(errors="replace")}

curl -X POST localhost:8000/covers/upload -F "cover=@cover.png;type=image/png"

{
  "filename": "cover.png",
  "content_type": "image/png",
  "size": 108,
  "headers": {
    "content-disposition": "form-data; name=\"cover\"; filename=\"cover.png\"",
    "content-type": "image/png"
  },
  "first_bytes": "89504e47",
  "bytes_read": 108
}

from typing import Annotated

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI(title="Bookshelf API")


@app.post("/books/{book_id}/sample")
async def optional_sample(book_id: int, sample: UploadFile | None = None) -> dict:
    if sample is None:
        return {"book_id": book_id, "message": "No sample sent"}
    return {"book_id": book_id, "filename": sample.filename}


@app.post("/books/{book_id}/photos")
async def many_photos(book_id: int, photos: list[UploadFile]) -> dict:
    return {"book_id": book_id, "filenames": [p.filename for p in photos]}


@app.post("/books/{book_id}/excerpts")
async def many_small_files(
    book_id: int,
    excerpts: Annotated[list[bytes], File(description="Short text excerpts")],
) -> dict:
    return {"book_id": book_id, "sizes": [len(e) for e in excerpts]}


@app.get("/upload-demo", response_class=HTMLResponse)
async def upload_demo() -> str:
    return """
<form action="/books/1/photos" enctype="multipart/form-data" method="post">
  <input name="photos" type="file" multiple accept="image/*">
  <input type="submit" value="Upload">
</form>
"""

curl -X POST localhost:8000/books/1/photos -F "photos=@a.jpg" -F "photos=@b.jpg"

{"book_id": 1, "filenames": ["a.jpg", "b.jpg"]}

import os
import uuid
from pathlib import Path
from typing import Annotated

import anyio
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status

UPLOAD_DIR = Path(os.environ.get("BOOKSHELF_UPLOAD_DIR", "uploads"))
MAX_COVER_BYTES = 2 * 1024 * 1024
CHUNK_SIZE = 64 * 1024
ALLOWED = {  # content type -> (extension, magic number)
    "image/png": (".png", b"\x89PNG\r\n\x1a\n"),
    "image/jpeg": (".jpg", b"\xff\xd8\xff"),
}

app = FastAPI(title="Bookshelf API")


@app.post("/books/{book_id}/cover", status_code=status.HTTP_201_CREATED)
async def upload_cover(
    book_id: int,
    cover: Annotated[UploadFile, File(description="PNG or JPEG, at most 2 MiB")],
    caption: Annotated[str | None, Form(max_length=200)] = None,
) -> dict:
    # 1. The declared type is only a hint from the client, so allow-list it AND check the real bytes.
    if cover.content_type not in ALLOWED:
        raise HTTPException(status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "Only PNG and JPEG covers are accepted")
    extension, magic = ALLOWED[cover.content_type]
    if not (await cover.read(len(magic))).startswith(magic):
        raise HTTPException(status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "File content does not match its type")
    await cover.seek(0)

    # 2. Never use the client's file name for the path: generate one.
    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
    stored_name = f"{book_id}-{uuid.uuid4().hex}{extension}"
    target = UPLOAD_DIR / stored_name

    # 3. Copy in chunks and stop as soon as the limit is exceeded.
    written = 0
    async with await anyio.open_file(target, "wb") as out:
        while chunk := await cover.read(CHUNK_SIZE):
            written += len(chunk)
            if written > MAX_COVER_BYTES:
                break
            await out.write(chunk)
    if written > MAX_COVER_BYTES:
        target.unlink(missing_ok=True)
        raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, f"Cover larger than {MAX_COVER_BYTES} bytes")
    return {"book_id": book_id, "stored_as": stored_name, "size": written, "caption": caption}

curl -i -F "cover=@cover.png;type=image/png" -F "caption=First edition" localhost:8000/books/7/cover

{"book_id": 7, "stored_as": "7-2cb2289a42dc40dea722678581b8e269.png", "size": 12, "caption": "First edition"}

from typing import Annotated

from fastapi import FastAPI, File, Form, UploadFile
from pydantic import BaseModel, Json

app = FastAPI(title="Bookshelf API")


class BookMeta(BaseModel):
    title: str
    isbn: str | None = None


# Broken on purpose: `meta` is a JSON body field, `token` is a form field.
@app.post("/books/broken")
async def create_book_broken(meta: BookMeta, token: Annotated[str, Form()]) -> dict:
    return {"title": meta.title}


# Working: the JSON travels as a string inside one multipart field and Pydantic parses it.
@app.post("/books/with-cover")
async def create_book(
    meta: Annotated[Json[BookMeta], Form(description="JSON document")],
    cover: UploadFile,
) -> dict:
    return {"title": meta.title, "isbn": meta.isbn, "cover": cover.filename}

curl -X POST localhost:8000/books/with-cover \
  -F 'meta={"title": "Dune", "isbn": "9780441172719"}' -F "cover=@cover.png;type=image/png"

{"title": "Dune", "isbn": "9780441172719", "cover": "cover.png"}

from fastapi import FastAPI, HTTPException, Request, UploadFile, status
from fastapi.responses import JSONResponse
from starlette.types import ASGIApp, Message, Receive, Scope, Send

MAX_BODY = 1 * 1024 * 1024  # 1 MiB


class LimitRequestBodySize:
    """Pure ASGI middleware: reject oversized bodies before FastAPI parses them."""

    def __init__(self, app: ASGIApp, max_bytes: int) -> None:
        self.app = app
        self.max_bytes = max_bytes

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        declared = dict(scope["headers"]).get(b"content-length")
        if declared is not None and declared.isdigit() and int(declared) > self.max_bytes:
            response = JSONResponse({"detail": "Request body too large"}, status.HTTP_413_CONTENT_TOO_LARGE)
            await response(scope, receive, send)  # rejected without reading the body
            return

        received = 0

        async def limited_receive() -> Message:
            nonlocal received
            message = await receive()
            if message["type"] == "http.request":
                received += len(message.get("body", b""))
                if received > self.max_bytes:  # chunked uploads have no Content-Length
                    raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, "Request body too large")
            return message

        await self.app(scope, limited_receive, send)


app = FastAPI(title="Bookshelf API")
app.add_middleware(LimitRequestBodySize, max_bytes=MAX_BODY)


@app.post("/books/{book_id}/cover-form")
async def upload_cover_form(book_id: int, cover: UploadFile) -> dict:
    return {"book_id": book_id, "size": cover.size}


@app.put("/books/{book_id}/cover-raw")
async def upload_cover_raw(book_id: int, request: Request) -> dict:
    """Raw upload (Content-Type: image/png, body = the file), consumed chunk by chunk."""
    received = 0
    async for chunk in request.stream():
        received += len(chunk)
        # ... write the chunk to disk or to an object store here ...
    return {"book_id": book_id, "size": received}

from typing import Annotated

from fastapi import FastAPI, Form, Query
from fastapi.responses import HTMLResponse

app = FastAPI(title="Bookshelf API")


@app.get("/search-page", response_class=HTMLResponse)
async def search_page() -> str:
    return """
<form action="/search" method="get">            <!-- fields go in the query string -->
  <input name="q"> <input type="submit" value="Search">
</form>
<form action="/books" method="post">            <!-- fields go in the body (urlencoded) -->
  <input name="title"> <input type="submit" value="Create">
</form>
"""


# Wrong: Form() on a GET operation reads the BODY, but a browser's GET form puts its fields in the URL.
@app.get("/search-wrong")
async def search_wrong(q: Annotated[str, Form()]) -> dict:
    return {"q": q}


# Right for GET: query parameters.
@app.get("/search")
async def search(q: Annotated[str, Query(min_length=1)]) -> dict:
    return {"q": q}


# Right for POST: form fields.
@app.post("/books")
async def create_book(title: Annotated[str, Form(min_length=1)]) -> dict:
    return {"title": title}

curl "localhost:8000/search-wrong?q=dune"

{"detail": [{"type": "missing", "loc": ["body", "q"], "msg": "Field required", "input": null}]}