File Uploads and Validation

Capsule overview

Your API handles JSON perfectly. But what happens when a user wants to attach a PDF to a task? Or upload their profile picture? Or import a CSV with data? Files don't travel as JSON — they travel as multipart/form-data, and FastAPI handles them with UploadFile.

UploadFile is FastAPI/Starlette's abstraction for uploaded files. It gives you access to the file's name, its MIME type, its size, and a file-like object to read its content. Combined with type and size validation, you can build upload endpoints that are safe and solid.

By the end of this capsule you'll know how to receive files with UploadFile, validate MIME type and size, save files to disk, combine an upload with JSON data using Form(), and handle multiple files in a single request.

UploadFile: the basics

Your first upload endpoint

from fastapi import FastAPI, UploadFile

app = FastAPI()


@app.post("/upload")
async def upload_file(file: UploadFile):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file.size
    }

That's all. FastAPI:

1. Detects that file is of type UploadFile
2. Expects multipart/form-data in the request
3. Parses the file and hands it to you as an object

Testing it with curl

curl -X POST http://localhost:8000/upload \
  -F "file=@document.pdf"

{
    "filename": "document.pdf",
    "content_type": "application/pdf",
    "size": 124532
}

Testing it from /docs

FastAPI generates a form in /docs with a file upload field. You can drag a file straight into it.

UploadFile properties

UploadFile methods

@app.post("/upload/read")
async def upload_read(file: UploadFile):
    content = await file.read()

    await file.seek(0)
    first_100 = await file.read(100)

    await file.close()

    return {
        "filename": file.filename,
        "total_bytes": len(content),
        "first_100_bytes": len(first_100)
    }

await file.read() — Reads all the content (bytes)

await file.read(n) — Reads n bytes

await file.seek(0) — Goes back to the start (to re-read)

await file.close() — Closes the temporary file

Saving files to disk

import os
from pathlib import Path
from fastapi import FastAPI, UploadFile

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)


@app.post("/upload/save")
async def save_file(file: UploadFile):
    file_path = UPLOAD_DIR / file.filename

    content = await file.read()
    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "filename": file.filename,
        "saved_to": str(file_path),
        "size_bytes": len(content)
    }

Saving in chunks (large files)

For large files, reading everything into memory can be a problem. Use chunks:

import shutil
from fastapi import FastAPI, UploadFile
from pathlib import Path

app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

CHUNK_SIZE = 1024 * 1024  # 1 MB


@app.post("/upload/chunked")
async def save_chunked(file: UploadFile):
    file_path = UPLOAD_DIR / file.filename
    total_size = 0

    with open(file_path, "wb") as f:
        while chunk := await file.read(CHUNK_SIZE):
            f.write(chunk)
            total_size += len(chunk)

    return {
        "filename": file.filename,
        "size_bytes": total_size,
        "chunks_used": (total_size // CHUNK_SIZE) + 1
    }

The walrus operator (:=) reads 1MB chunks until there are no bytes left.

File validation

Validating the MIME type

from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

ALLOWED_TYPES = {
    "image/jpeg",
    "image/png",
    "image/gif",
    "image/webp",
}


@app.post("/upload/image")
async def upload_image(file: UploadFile):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=400,
            detail={
                "error": "File type not allowed",
                "received": file.content_type,
                "allowed": list(ALLOWED_TYPES)
            }
        )

    content = await file.read()
    return {
        "filename": file.filename,
        "type": file.content_type,
        "size_bytes": len(content)
    }

Warning: content_type comes from the client and can be forged. For real validation, read the file's first bytes (magic bytes) or use a library like python-magic.

Validating size

from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

MAX_SIZE = 5 * 1024 * 1024  # 5 MB


@app.post("/upload/limited")
async def upload_limited(file: UploadFile):
    content = await file.read()

    if len(content) > MAX_SIZE:
        raise HTTPException(
            status_code=413,
            detail={
                "error": "File too large",
                "max_size_mb": MAX_SIZE / (1024 * 1024),
                "received_mb": round(len(content) / (1024 * 1024), 2)
            }
        )

    return {"filename": file.filename, "size_mb": round(len(content) / (1024 * 1024), 2)}

Validating the extension

from pathlib import Path
from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

ALLOWED_EXTENSIONS = {".pdf", ".doc", ".docx", ".txt", ".csv"}


@app.post("/upload/docs")
async def upload_document(file: UploadFile):
    extension = Path(file.filename).suffix.lower()

    if extension not in ALLOWED_EXTENSIONS:
        raise HTTPException(
            status_code=400,
            detail={
                "error": "Extension not allowed",
                "received": extension,
                "allowed": list(ALLOWED_EXTENSIONS)
            }
        )

    content = await file.read()
    return {"filename": file.filename, "extension": extension, "size": len(content)}

A reusable validation function

Combining all the validations:

from pathlib import Path
from fastapi import UploadFile, HTTPException

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB
ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".pdf"}


async def validate_upload(
    file: UploadFile,
    max_size: int = MAX_FILE_SIZE,
    allowed_types: set[str] = ALLOWED_TYPES,
    allowed_extensions: set[str] = ALLOWED_EXTENSIONS
) -> bytes:
    extension = Path(file.filename).suffix.lower()
    if extension not in allowed_extensions:
        raise HTTPException(400, detail=f"Extension {extension} not allowed")

    if file.content_type not in allowed_types:
        raise HTTPException(400, detail=f"Type {file.content_type} not allowed")

    content = await file.read()
    if len(content) > max_size:
        raise HTTPException(413, detail=f"File exceeds {max_size / (1024*1024):.1f} MB")

    return content

@app.post("/upload/validated")
async def upload_validated(file: UploadFile):
    content = await validate_upload(file)

    file_path = UPLOAD_DIR / file.filename
    with open(file_path, "wb") as f:
        f.write(content)

    return {"filename": file.filename, "size": len(content), "saved": True}

Combining File Upload with Form Data

What happens when you want to send a file and additional data? You can't use a JSON body and a file upload at the same time. The solution is Form():

from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from pathlib import Path

app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)


@app.post("/tasks/{task_id}/attachments")
async def add_attachment(
    task_id: int,
    file: UploadFile = File(...),
    description: str = Form(default=""),
    category: str = Form(default="general")
):
    content = await file.read()

    file_path = UPLOAD_DIR / f"task_{task_id}_{file.filename}"
    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "task_id": task_id,
        "attachment": {
            "filename": file.filename,
            "description": description,
            "category": category,
            "size_bytes": len(content),
            "content_type": file.content_type,
            "path": str(file_path)
        }
    }

File(...) — Explicitly marks it as a file field (required)

Form(default="") — A text field that comes in the form data, not in JSON

Testing it with curl

curl -X POST http://localhost:8000/tasks/1/attachments \
  -F "file=@spec.pdf" \
  -F "description=Project specification" \
  -F "category=documentation"

Important: When you use File() or Form(), the body can NOT be JSON (-H "Content-Type: application/json"). Everything travels as multipart/form-data.

Multiple files

A list of files

from fastapi import FastAPI, UploadFile, File

app = FastAPI()


@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    results = []
    for file in files:
        content = await file.read()
        results.append({
            "filename": file.filename,
            "content_type": file.content_type,
            "size_bytes": len(content)
        })

    return {
        "files_received": len(results),
        "total_size": sum(r["size_bytes"] for r in results),
        "files": results
    }

curl -X POST http://localhost:8000/upload/multiple \
  -F "files=@file1.pdf" \
  -F "files=@file2.png" \
  -F "files=@file3.txt"

Multiple files with different fields

@app.post("/profile/complete")
async def complete_profile(
    avatar: UploadFile = File(...),
    resume: UploadFile = File(...),
    bio: str = Form(default="")
):
    avatar_bytes = await avatar.read()
    resume_bytes = await resume.read()

    return {
        "avatar": {"name": avatar.filename, "size": len(avatar_bytes)},
        "resume": {"name": resume.filename, "size": len(resume_bytes)},
        "bio": bio
    }

Generating unique names

In production, never save files with the user's original name (there can be collisions, special characters, or malicious names):

import uuid
from pathlib import Path
from datetime import datetime


def generate_unique_filename(original_filename: str) -> str:
    extension = Path(original_filename).suffix.lower()
    unique_id = uuid.uuid4().hex[:12]
    timestamp = datetime.now().strftime("%Y%m%d")
    return f"{timestamp}_{unique_id}{extension}"

@app.post("/upload/safe")
async def safe_upload(file: UploadFile):
    safe_name = generate_unique_filename(file.filename)
    file_path = UPLOAD_DIR / safe_name

    content = await file.read()
    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "original_name": file.filename,
        "stored_as": safe_name,
        "path": str(file_path)
    }

Serving saved files

Use FileResponse to serve saved files:

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path

app = FastAPI()
UPLOAD_DIR = Path("uploads")


@app.get("/files/{filename}")
async def get_file(filename: str):
    file_path = UPLOAD_DIR / filename

    if not file_path.exists():
        raise HTTPException(404, detail=f"File {filename} not found")

    if not file_path.is_relative_to(UPLOAD_DIR):
        raise HTTPException(403, detail="Access denied")

    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream"
    )

is_relative_to() — Prevents path traversal attacks (e.g.: ../../../etc/passwd).

Exercises

Exercise 1: Upload with full validation

Create a POST /upload/image endpoint that:

• Only accepts images (jpeg, png, webp)
• Has a 2 MB limit
• Saves with a unique name
• Returns the URL to access the file

import uuid
from pathlib import Path
from fastapi import FastAPI, UploadFile, HTTPException
from fastapi.responses import FileResponse

app = FastAPI()
UPLOAD_DIR = Path("uploads/images")
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp"}
MAX_IMAGE_SIZE = 2 * 1024 * 1024


@app.post("/upload/image")
async def upload_image(file: UploadFile):
    if file.content_type not in ALLOWED_IMAGE_TYPES:
        raise HTTPException(400, detail=f"Images only: {ALLOWED_IMAGE_TYPES}")

    content = await file.read()
    if len(content) > MAX_IMAGE_SIZE:
        raise HTTPException(413, detail="2 MB maximum")

    ext = Path(file.filename).suffix.lower()
    unique_name = f"{uuid.uuid4().hex[:12]}{ext}"
    file_path = UPLOAD_DIR / unique_name

    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "filename": unique_name,
        "original_name": file.filename,
        "size_kb": round(len(content) / 1024, 1),
        "url": f"/images/{unique_name}"
    }


@app.get("/images/{filename}")
async def get_image(filename: str):
    file_path = UPLOAD_DIR / filename
    if not file_path.exists():
        raise HTTPException(404, detail="Image not found")
    return FileResponse(file_path)

Exercise 2: A CSV parser endpoint

Create an endpoint that takes a CSV, parses it, and returns the data as JSON. Validate that it's type text/csv and that it has at least a header.

import csv
import io
from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()


@app.post("/upload/csv")
async def parse_csv(file: UploadFile):
    if file.content_type not in {"text/csv", "application/csv"}:
        raise HTTPException(400, detail="CSV files only")

    content = await file.read()
    text = content.decode("utf-8")
    reader = csv.DictReader(io.StringIO(text))

    rows = list(reader)
    if not rows:
        raise HTTPException(400, detail="Empty CSV or no data")

    return {
        "filename": file.filename,
        "columns": list(rows[0].keys()),
        "row_count": len(rows),
        "data": rows[:10],
        "truncated": len(rows) > 10
    }

Exercise 3: Multiple upload with a summary

Create an endpoint that takes up to 5 files, validates each one (1MB max), and returns a summary with the total bytes, the unique types, and the accepted/rejected files.

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

MAX_FILES = 5
MAX_SIZE_PER_FILE = 1 * 1024 * 1024


@app.post("/upload/batch")
async def upload_batch(files: list[UploadFile] = File(...)):
    if len(files) > MAX_FILES:
        return {"error": f"{MAX_FILES} files maximum, you sent {len(files)}"}

    accepted = []
    rejected = []

    for file in files:
        content = await file.read()
        info = {
            "filename": file.filename,
            "content_type": file.content_type,
            "size_bytes": len(content)
        }

        if len(content) > MAX_SIZE_PER_FILE:
            info["reason"] = "Exceeds 1 MB"
            rejected.append(info)
        else:
            accepted.append(info)

    unique_types = set(f["content_type"] for f in accepted + rejected)

    return {
        "total_files": len(files),
        "accepted": len(accepted),
        "rejected": len(rejected),
        "total_bytes_accepted": sum(f["size_bytes"] for f in accepted),
        "unique_types": list(unique_types),
        "files_accepted": accepted,
        "files_rejected": rejected
    }

Exercise 4: File + Form + Path params

Create a POST /projects/{project_id}/documents endpoint that takes:

• project_id (path param)
• file (UploadFile)
• title (Form, required)
• tags (Form, optional, comma-separated)

Save the file and return complete metadata.

import uuid
from pathlib import Path
from fastapi import FastAPI, UploadFile, File, Form

app = FastAPI()
UPLOAD_DIR = Path("uploads/projects")
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)


@app.post("/projects/{project_id}/documents")
async def upload_document(
    project_id: int,
    file: UploadFile = File(...),
    title: str = Form(...),
    tags: str = Form(default="")
):
    project_dir = UPLOAD_DIR / str(project_id)
    project_dir.mkdir(exist_ok=True)

    ext = Path(file.filename).suffix
    safe_name = f"{uuid.uuid4().hex[:8]}{ext}"
    file_path = project_dir / safe_name

    content = await file.read()
    with open(file_path, "wb") as f:
        f.write(content)

    tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else []

    return {
        "project_id": project_id,
        "document": {
            "id": uuid.uuid4().hex[:8],
            "title": title,
            "original_filename": file.filename,
            "stored_as": safe_name,
            "content_type": file.content_type,
            "size_bytes": len(content),
            "tags": tag_list,
            "path": str(file_path)
        }
    }

Troubleshooting

422 Unprocessable Entity when uploading a file

python-multipart is probably missing. Install it: pip install python-multipart.

content_type is None

Some clients don't send the file's Content-Type. Use the extension as a fallback:

content_type = file.content_type or "application/octet-stream"

file.size returns None

In some versions of Starlette, size isn't available until you read the file. Read it first: content = await file.read() and use len(content).

Large files eat all the memory

Use the chunk pattern instead of await file.read(). Read in 1MB blocks with while chunk := await file.read(CHUNK_SIZE).

Path traversal: the filename contains ../

Never use file.filename directly to build paths. Generate unique names with uuid or sanitize with Path(file.filename).name.

Resources

1. FastAPI Request Files — The official docs
2. FastAPI Request Forms and Files — Combining Form + File
3. Starlette UploadFile — The base API
4. python-multipart — The multipart parser
5. FileResponse — Serving files
6. OWASP File Upload — Security

What's next?

In Capsule 05 (the module project), you'll integrate WebSockets and file uploads into your Task Manager API. You'll build a real-time notification system that tells every connected client when tasks are created or completed, and you'll add the ability to attach files to each task.