Chapter 19: File Uploads — Image Processing, Validation, and Storage

By /

Let’s Start With a Real Problem

You add a “profile picture” feature to your app. It works great in your own testing. Then, within a week: someone uploads a 40MB image and your server’s disk usage spikes. Someone else renames a .exe file to photo.jpg and uploads it. A third person uploads a genuinely corrupted file that crashes your image-processing code. None of these are exotic attacks — they’re the completely predictable, everyday reality of accepting files from the public internet.

This chapter is about handling file uploads the way a production API actually needs to: validating what comes in, processing images safely, and storing everything in an organized, predictable way.

What You’ll Learn

  • How to accept file uploads in FastAPI using UploadFile
  • How to validate file type and file size before trusting anything about the upload
  • How to safely open and verify an image using Pillow, rejecting corrupted or fake files
  • How to resize images and store them with unique, collision-proof filenames
  • How to serve uploaded files back out through your API
  • Theory + hands-on assignment questions to lock in what you’ve learned

Theory: Why File Uploads Need Extra Care

A file upload is fundamentally different from a JSON request body — Pydantic can’t validate the contents of a file the way it validates a string or a number. That means you are responsible for checking three things, in this order, before doing anything else with an uploaded file:

  1. Type — is this actually the kind of file you expect? (Never trust the filename extension alone — it’s just text the client typed.)
  2. Size — is it within a reasonable limit? An unbounded upload endpoint is an easy way for someone to fill up your server’s disk.
  3. Integrity — for images specifically, can it actually be opened and decoded as a real image, or is it corrupted or not an image at all?

⚠️ Never trust Content-Type alone. A malicious client can set any Content-Type header they like, regardless of what the file actually contains. For images, the only reliable check is genuinely trying to open and decode the file with an imaging library — which is exactly what Program 2 does with Pillow.

Program 1: Basic Upload with Type and Size Validation

Program Objective

Build a small, standalone FastAPI project with an upload endpoint that validates an incoming file’s content type and size, saves it to disk with a unique filename, and serves it back through a static file URL.

Step-by-Step Implementation

Step 1: Create the project

uv init file-upload-basics-demo
cd file-upload-basics-demo
uv add fastapi --extra standard

💡 python-multipart (required for handling file uploads) is already included via the standard extra, so no separate install is needed.

Step 2: Create the project structure

mkdir -p app/routers uploads
touch app/__init__.py app/routers/__init__.py
file-upload-basics-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   └── routers/
│       ├── __init__.py
│       └── uploads.py
├── uploads/
├── pyproject.toml
└── uv.lock

Step 3: Create each file with the code below. Step 4: Run and test.

Complete Working Source Code

app/routers/uploads.py

import uuid
from pathlib import Path

from fastapi import APIRouter, HTTPException, UploadFile, status

router = APIRouter(prefix="/upload", tags=["Uploads"])

UPLOAD_DIR = Path("uploads")
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}
MAX_FILE_SIZE_BYTES = 5 * 1024 * 1024  # 5 MB


@router.post("/image", status_code=status.HTTP_201_CREATED)
async def upload_image(file: UploadFile):
    # 1. Validate content type
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Unsupported file type: {file.content_type}. Allowed: {', '.join(ALLOWED_CONTENT_TYPES)}.",
        )

    # 2. Read the file into memory and validate size
    contents = await file.read()
    if len(contents) > MAX_FILE_SIZE_BYTES:
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail=f"File too large. Maximum allowed size is {MAX_FILE_SIZE_BYTES // (1024 * 1024)} MB.",
        )

    # 3. Build a unique, collision-proof filename
    file_extension = Path(file.filename).suffix
    unique_filename = f"{uuid.uuid4().hex}{file_extension}"
    destination_path = UPLOAD_DIR / unique_filename

    # 4. Save the file to disk
    with open(destination_path, "wb") as f:
        f.write(contents)

    return {
        "original_filename": file.filename,
        "saved_as": unique_filename,
        "content_type": file.content_type,
        "size_bytes": len(contents),
        "url": f"/uploads/{unique_filename}",
    }

app/main.py

from pathlib import Path

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from app.routers import uploads

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

app = FastAPI(title="File Upload Basics Demo", version="1.0.0")

app.include_router(uploads.router)
app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads")


@app.get("/", tags=["Root"])
def read_root():
    return {"message": "Welcome to File Upload Basics Demo"}

Detailed Code Explanation

  • file: UploadFile — FastAPI recognizes this type and automatically parses incoming multipart/form-data, giving you a file-like object with .filename, .content_type, and async methods like .read(). UploadFile is preferred over the older bytes-based approach because it streams the file rather than loading the entire thing into memory immediately, though we do call .read() here to get the full contents for validation and saving.
  • Step 1 — content type check happens first and is cheap: reject obviously wrong file types before spending any effort reading or processing the file at all.
  • Step 2 — size check happens after reading the bytes, since we need the actual byte count to check against MAX_FILE_SIZE_BYTES. 413 Request Entity Too Large is the semantically correct status code for this specific failure — more precise than a generic 400.
  • Step 3 — the unique filename is the detail most beginners miss: if you saved the file using its original filename, two different users uploading photo.jpg would silently overwrite each other’s files. uuid.uuid4().hex generates a random, effectively collision-proof identifier, and we preserve the original file extension so the saved file still opens correctly.
  • app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads") in main.py is what actually makes uploaded files retrievable — any file saved into the uploads/ folder becomes accessible at http://127.0.0.1:8000/uploads/<filename>, without needing a dedicated “download” endpoint.
  • UPLOAD_DIR.mkdir(exist_ok=True) ensures the folder exists before the app tries to save anything into it — exist_ok=True means this won’t raise an error if the folder is already there from a previous run.

Execution & Testing

uv run fastapi dev app/main.py
  1. Open http://127.0.0.1:8000/docs, expand POST /upload/image, click Try it out, and choose a real .jpg or .png file from your computer. Execute, and confirm you get back a response with a url field.
  2. Paste that URL (e.g. http://127.0.0.1:8000/uploads/<generated-name>.jpg) into a new browser tab — confirm the actual image loads.
  3. Try uploading a .txt or .pdf file — confirm you get a 400 explaining the content type isn’t allowed.
  4. Try uploading a genuinely large file (over 5MB, e.g. a high-resolution photo) — confirm you get a 413 explaining it’s too large.
  5. Upload the same image file twice in a row — check your uploads/ folder and confirm two different filenames were created, proving the collision protection works.

Program 2: Processing and Storing a Profile Picture

Program Objective

Build a small, standalone FastAPI project with a protected POST /users/me/avatar endpoint that validates an uploaded image using Pillow, resizes it to a consistent size, saves it under a predictable per-user filename, and updates the logged-in user’s avatar_url in the database.

Step-by-Step Implementation

Step 1: Create the project

uv init avatar-upload-demo
cd avatar-upload-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy
uv add "passlib[bcrypt]"
uv add "python-jose[cryptography]"
uv add pillow

Step 2: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers uploads/avatars
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py
avatar-upload-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   ├── security.py
│   │   └── dependencies.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── token.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── avatar_service.py
│   └── routers/
│       ├── __init__.py
│       ├── users.py
│       └── auth.py
├── uploads/
│   └── avatars/
├── pyproject.toml
└── uv.lock

Step 3: Create each file with the code below. Step 4: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Avatar Upload Demo"
    app_version: str = "1.0.0"
    database_url: str = "sqlite:///./app.db"
    jwt_secret_key: str = "a-very-insecure-demo-secret-change-this"
    jwt_algorithm: str = "HS256"
    access_token_expire_minutes: int = 30


settings = Settings()

app/core/security.py

from datetime import datetime, timedelta, timezone

from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plain_password: str) -> str:
    return pwd_context.hash(plain_password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


def create_access_token(subject: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    to_encode = {"sub": subject, "exp": expire}
    return jwt.encode(to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)


def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
    except JWTError as exc:
        raise ValueError("Invalid or expired token.") from exc

app/core/dependencies.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.core.security import decode_token
from app.db.database import get_db
from app.models.user import User
from app.services.user_service import get_user_by_username

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")


def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials.",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = decode_token(token)
    except ValueError:
        raise credentials_exception

    username = payload.get("sub")
    if username is None:
        raise credentials_exception

    user = get_user_by_username(db, username)
    if user is None:
        raise credentials_exception

    return user

app/db/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/user.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(30), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    avatar_url: Mapped[str | None] = mapped_column(String(255), default=None)

app/schemas/user.py

from typing import Annotated

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserCreate(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=30)]
    email: Annotated[EmailStr, Field()]
    password: Annotated[str, Field(min_length=8, max_length=64)]


class UserOut(BaseModel):
    id: int
    username: str
    email: EmailStr
    avatar_url: str | None

    model_config = ConfigDict(from_attributes=True)

app/schemas/token.py

from pydantic import BaseModel


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

app/services/user_service.py

from sqlalchemy.orm import Session

from app.core.security import hash_password
from app.models.user import User
from app.schemas.user import UserCreate


def create_user(db: Session, user_in: UserCreate) -> User:
    new_user = User(
        username=user_in.username,
        email=user_in.email,
        hashed_password=hash_password(user_in.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


def get_user_by_username(db: Session, username: str) -> User | None:
    return db.query(User).filter(User.username == username).first()

app/services/avatar_service.py

import io
from pathlib import Path

from fastapi import HTTPException, UploadFile, status
from PIL import Image, UnidentifiedImageError
from sqlalchemy.orm import Session

from app.models.user import User

AVATAR_DIR = Path("uploads/avatars")
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}
MAX_FILE_SIZE_BYTES = 5 * 1024 * 1024  # 5 MB
AVATAR_SIZE = (200, 200)


async def process_and_save_avatar(db: Session, user: User, file: UploadFile) -> User:
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Unsupported file type: {file.content_type}.",
        )

    contents = await file.read()
    if len(contents) > MAX_FILE_SIZE_BYTES:
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail="File too large. Maximum allowed size is 5 MB.",
        )

    # Verify the file is genuinely a valid, undamaged image.
    try:
        image = Image.open(io.BytesIO(contents))
        image.verify()
    except UnidentifiedImageError:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="The uploaded file is not a valid image.",
        )

    # image.verify() invalidates the file object, so we must re-open it to actually use it.
    image = Image.open(io.BytesIO(contents))
    image = image.convert("RGB")  # normalizes PNG/WEBP transparency before saving as JPEG
    image.thumbnail(AVATAR_SIZE)

    destination_path = AVATAR_DIR / f"{user.id}.jpg"
    image.save(destination_path, format="JPEG", quality=85)

    user.avatar_url = f"/uploads/avatars/{user.id}.jpg"
    db.commit()
    db.refresh(user)
    return user

app/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.core.security import create_access_token, verify_password
from app.db.database import get_db
from app.schemas.token import TokenResponse
from app.services.user_service import get_user_by_username

router = APIRouter(prefix="/auth", tags=["Auth"])


@router.post("/login", response_model=TokenResponse)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = get_user_by_username(db, form_data.username)

    if user is None or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password.")

    return TokenResponse(access_token=create_access_token(subject=user.username))

app/routers/users.py

from fastapi import APIRouter, Depends, UploadFile, status
from sqlalchemy.orm import Session

from app.core.dependencies import get_current_user
from app.db.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserOut
from app.services.avatar_service import process_and_save_avatar
from app.services.user_service import create_user

router = APIRouter(prefix="/users", tags=["Users"])


@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user_in: UserCreate, db: Session = Depends(get_db)):
    return create_user(db, user_in)


@router.get("/me", response_model=UserOut)
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user


@router.post("/me/avatar", response_model=UserOut)
async def upload_avatar(
    file: UploadFile,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    return await process_and_save_avatar(db, current_user, file)

app/main.py

from pathlib import Path

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from app.core.config import settings
from app.db.database import Base, engine
from app.models import user  # noqa: F401
from app.routers import auth, users

Path("uploads/avatars").mkdir(parents=True, exist_ok=True)

Base.metadata.create_all(bind=engine)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(users.router)
app.include_router(auth.router)
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

  • User.avatar_url: Mapped[str | None] — nullable, since a freshly registered user hasn’t uploaded a picture yet. UserOut reflects this with avatar_url: str | None, so the API honestly represents “no avatar yet” as null, not an empty string or a broken link.
  • The real integrity check happens here:
  image = Image.open(io.BytesIO(contents))
  image.verify()

Image.open(...) alone doesn’t fully decode the image — it can succeed even on some malformed files. .verify() forces Pillow to actually check the file’s internal structure is valid, raising UnidentifiedImageError (or similar) if it’s corrupted or not a real image at all — this is exactly the check that a spoofed Content-Type header can’t fool.

  • The re-open after .verify() is not a mistake — it’s a documented Pillow requirement. Calling .verify() leaves the image object unusable for further processing, so we deliberately create a fresh Image.open(...) call afterward to actually work with the image data.
  • image.convert("RGB") — PNG and WEBP files can include a transparency channel that JPEG doesn’t support. Converting to RGB before saving avoids a crash or a strange-looking output when we save everything as .jpg for consistency.
  • image.thumbnail(AVATAR_SIZE) resizes the image to fit within 200x200 pixels while preserving its aspect ratio — unlike a hard resize, .thumbnail() won’t stretch or distort a non-square photo.
  • destination_path = AVATAR_DIR / f"{user.id}.jpg" — notice this filename is predictable, not random, unlike Program 1’s upload. That’s deliberate: every time this specific user uploads a new avatar, it overwrites their previous one at the exact same path, so avatar_url never needs to change once it’s set, and old avatar files don’t pile up on disk.
  • process_and_save_avatar is called from a protected route (Depends(get_current_user)) — combining everything from Chapters 15–18: only a logged-in user can upload an avatar, and it always updates their own record (current_user, never an ID taken from the request).

Execution & Testing

uv run fastapi dev app/main.py
  1. Register a user and log in via POST /auth/login to get a token; Authorize in Swagger UI using that token.
  2. Call POST /users/me/avatar, uploading a real photo. Confirm the response’s avatar_url field looks like /uploads/avatars/1.jpg.
  3. Open http://127.0.0.1:8000/uploads/avatars/1.jpg directly in your browser — confirm you see a resized, square-fitted version of your photo.
  4. Upload a different photo through the same endpoint again — confirm it overwrites the same file (same URL, new image content), rather than creating a second file.
  5. Try uploading a renamed non-image file (e.g., rename a .txt file to fake.jpg and force its content type, or simply try uploading any corrupted image file you have) — confirm you get a clean 400 The uploaded file is not a valid image, proving Pillow’s .verify() step is doing real work, not just checking the file extension.

Tips & Instructions for Developers

  • ✅ Always validate content type, size, and (for images) integrity — in that order, cheapest checks first — before doing any real processing.
  • ✅ Use uuid4()-based filenames for general uploads, but predictable, per-owner filenames (like {user_id}.jpg) when a resource should only ever have one current file, like an avatar.
  • ✅ Always convert images to a consistent format and color mode (like RGB JPEG) before saving, to avoid format-specific surprises later.
  • ✅ Store uploaded files outside your application code directory, and serve them through a dedicated static route or a proper object storage service in production — never assume local disk storage scales indefinitely.
  • ✅ Tie file uploads to the authenticated user wherever ownership matters, exactly like the ownership patterns from Chapter 18.
  • ❌ Don’t trust file.content_type as your only validation — it’s client-supplied and easy to fake; for images, Pillow’s .verify() is your real safety net.
  • ❌ Don’t save uploaded files using their original filename — it invites both overwrites/collisions and, on some systems, path traversal risks if the filename isn’t sanitized.

Official FastAPI Documentation References

Theory Questions

  1. Why is file.content_type not sufficient on its own to guarantee an uploaded file is genuinely the type it claims to be?
  2. Why does validating file size happen after reading the file’s contents, rather than before?
  3. Why is Image.open(...).verify() a stronger check than simply confirming the file opens without an error?
  4. Why does the avatar upload use a predictable filename ({user_id}.jpg) while the general upload in Program 1 uses a random one?
  5. Why does the avatar endpoint use current_user.id for the filename and database update, rather than accepting a user_id from the request?

Practical Assignment

  1. Add a dimensions check: Starting from Program 1’s project, after validating type and size, open the file with Pillow and reject any image smaller than 100x100 pixels, returning a clear 400 error.
  2. Generate a thumbnail alongside the original: Starting from Program 1’s project, in addition to saving the original uploaded image, also generate and save a 150x150 thumbnail version alongside it (e.g. <uuid>_thumb.jpg), and return both URLs in the response.
  3. Add avatar deletion: Starting from Program 2’s project, add a DELETE /users/me/avatar endpoint that removes the user’s avatar file from disk (if it exists) and sets avatar_url back to None in the database.

Summary

In this chapter, you learned:

  • Why file uploads require validation that Pydantic alone can’t provide
  • How to check content type and size before trusting an uploaded file
  • How to use Pillow’s .verify() to catch corrupted or fake image files that a spoofed Content-Type header can’t hide
  • How to resize images consistently and store them with either collision-proof or predictable filenames, depending on the use case
  • How to tie file uploads to the authenticated user, combining everything from the authentication and authorization chapters
Scroll to Top