FastAPI handles file uploads through the UploadFile class — a wrapper around the uploaded file that provides metadata (filename, content type) and streaming access to the file contents. Unlike reading the entire file into memory with bytes, UploadFile streams the data, making it safe for large files. Combine UploadFile with Form fields to accept mixed requests containing both files and structured data — the file upload for a user avatar and the display name in the same request, for example.
Single File Upload
from fastapi import FastAPI, UploadFile, File, HTTPException, status
from pathlib import Path
import shutil, uuid
app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload", status_code=201)
async def upload_file(file: UploadFile = File(...)):
"""Accept any file upload and save to disk."""
# Validate file type by content-type header
allowed_types = {"image/jpeg", "image/png", "image/webp", "application/pdf"}
if file.content_type not in allowed_types:
raise HTTPException(
status_code = 415,
detail = f"Unsupported file type: {file.content_type}",
)
# Validate file size (read first chunk to check early)
MAX_SIZE_BYTES = 10 * 1024 * 1024 # 10 MB
content = await file.read() # reads entire file into memory
if len(content) > MAX_SIZE_BYTES:
raise HTTPException(413, f"File too large (max {MAX_SIZE_BYTES // 1024 // 1024} MB)")
# Generate safe unique filename (never trust the original name)
suffix = Path(file.filename).suffix.lower() if file.filename else ".bin"
filename = f"{uuid.uuid4()}{suffix}"
dest = UPLOAD_DIR / filename
# Write to disk
with open(dest, "wb") as f:
f.write(content)
return {
"filename": filename,
"original_name": file.filename,
"size": len(content),
"url": f"/uploads/{filename}",
}
UploadFile provides streaming access via await file.read() (reads all bytes) or await file.read(chunk_size) (streaming chunks). For small files (images, documents under ~10 MB), reading all at once is fine. For large files (videos, datasets), use chunked streaming to avoid loading gigabytes into memory: while chunk := await file.read(64 * 1024): dest_file.write(chunk). After file.read(), call await file.seek(0) if you need to read the file contents again (the read pointer is at the end).file.filename directly as the path to save the file. A malicious client can send a filename like ../../etc/passwd or ../config.py (path traversal attack). Always generate a new filename using uuid.uuid4() and only use the client’s filename for the extension (after validating it is a safe extension). Restrict saved files to a specific directory using Path(UPLOAD_DIR / filename).resolve() and verify the resolved path starts with UPLOAD_DIR.resolve().file.content_type for security-critical type validation. The content type is provided by the client and can be spoofed — a malicious user can upload a PHP script with content_type="image/jpeg". For security, use a library like python-magic to detect the actual file type from the file’s magic bytes (the first few bytes of the file content), in addition to checking the declared content type.Mixed File and Form Data
from fastapi import UploadFile, File, Form, Depends
from sqlalchemy.orm import Session
@app.post("/users/me/avatar", response_model=UserResponse)
async def upload_avatar(
avatar: UploadFile = File(...),
display_name: str = Form(default=None, max_length=100),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""
Upload a profile avatar with optional display name update.
Content-Type must be multipart/form-data (set by browser/client automatically).
"""
# Validate image type
if avatar.content_type not in ("image/jpeg", "image/png", "image/webp"):
raise HTTPException(415, "Avatar must be JPEG, PNG, or WebP")
content = await avatar.read()
if len(content) > 5 * 1024 * 1024: # 5 MB max for avatars
raise HTTPException(413, "Avatar must be under 5 MB")
# Save file
filename = f"avatars/{current_user.id}_{uuid.uuid4()}.jpg"
save_path = UPLOAD_DIR / filename
save_path.parent.mkdir(parents=True, exist_ok=True)
with open(save_path, "wb") as f:
f.write(content)
# Update user profile
profile = db.scalars(select(UserProfile).where(UserProfile.user_id == current_user.id)).first()
if profile:
profile.avatar_url = f"/uploads/{filename}"
if display_name:
current_user.name = display_name
db.flush()
return current_user
Multiple File Upload
from fastapi import UploadFile, File
from typing import Annotated
@app.post("/upload/multiple")
async def upload_multiple(
files: Annotated[list[UploadFile], File(description="Multiple files")],
):
"""Accept up to 10 files in a single request."""
if len(files) > 10:
raise HTTPException(400, "Maximum 10 files per request")
results = []
for file in files:
content = await file.read()
filename = f"{uuid.uuid4()}{Path(file.filename or '').suffix}"
(UPLOAD_DIR / filename).write_bytes(content)
results.append({
"filename": filename,
"original": file.filename,
"size": len(content),
})
return {"uploaded": len(results), "files": results}
Common Mistakes
Mistake 1 — Using client filename directly as save path
❌ Wrong — path traversal vulnerability:
dest = UPLOAD_DIR / file.filename # file.filename = "../../etc/passwd" → DISASTER
✅ Correct — generate safe UUID-based filename:
safe_name = f"{uuid.uuid4()}{Path(file.filename or '').suffix.lower()}"
dest = UPLOAD_DIR / safe_name # ✓ safe
Mistake 2 — Reading the entire file twice without seeking back
❌ Wrong — second read returns empty bytes:
content = await file.read() # reads all bytes
await validate_image(file) # file.read() inside returns b"" — already at end!
✅ Correct — seek back or pass content directly:
content = await file.read()
await file.seek(0) # ✓ reset read pointer
await validate_image(file) # now reads from the start again
Mistake 3 — Trusting content_type without magic byte validation
❌ Wrong — attacker uploads PHP with content_type=”image/jpeg”:
if file.content_type == "image/jpeg":
save(file) # saves malicious PHP script as "image"!
✅ Correct — validate magic bytes:
import imghdr
img_type = imghdr.what(None, h=content[:32])
if img_type not in ("jpeg", "png", "webp"):
raise HTTPException(415, "Invalid image format") # ✓ checks actual content
Quick Reference
| Task | Code |
|---|---|
| Single file | file: UploadFile = File(...) |
| Optional file | file: UploadFile | None = File(default=None) |
| Multiple files | files: list[UploadFile] = File(...) |
| File + form fields | Mix UploadFile = File(...) and str = Form(...) |
| Read all bytes | content = await file.read() |
| Stream chunks | while chunk := await file.read(65536): ... |
| Safe filename | f"{uuid.uuid4()}{Path(file.filename).suffix}" |