python 43 lines · 8 steps

Safe avatar uploads in FastAPI

A file upload endpoint that validates content type, streams to disk in chunks, and enforces a size cap while cleaning up on failure.

Explained by highlit

FastAPI Intermediate file-upload validation streaming error-handling cleanup

1import shutil
2import uuid
3from pathlib import Path
4 
5from fastapi import APIRouter, File, HTTPException, UploadFile, status
6 
7router = APIRouter(prefix="/uploads", tags=["uploads"])
8 
9UPLOAD_DIR = Path("media/avatars")
10ALLOWED_TYPES = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}
11MAX_BYTES = 5 * 1024 * 1024
12 
13 
14@router.post("/avatar", status_code=status.HTTP_201_CREATED)
15async def upload_avatar(file: UploadFile = File(...)):
16    extension = ALLOWED_TYPES.get(file.content_type)
17    if extension is None:
18        raise HTTPException(
19            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
20            detail=f"Unsupported content type: {file.content_type}",
21        )
22 
23    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
24    destination = UPLOAD_DIR / f"{uuid.uuid4().hex}{extension}"
25 
26    size = 0
27    try:
28        with destination.open("wb") as buffer:
29            while chunk := await file.read(1024 * 1024):
30                size += len(chunk)
31                if size > MAX_BYTES:
32                    raise HTTPException(
33                        status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
34                        detail="File exceeds the 5 MB limit",
35                    )
36                buffer.write(chunk)
37    except HTTPException:
38        destination.unlink(missing_ok=True)
39        raise
40    finally:
41        await file.close()
42 
43    return {"filename": destination.name, "content_type": file.content_type, "size": size}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1Streaming uploads chunk-by-chunk keeps memory bounded no matter how large the file is.
2Enforce size limits during the write, not before, since a client's declared length can't be trusted.
3Clean up partially written files on failure so a rejected upload never leaves junk on disk.

Related explainers

python

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

Parallel file downloads with a thread pool

concurrency thread-pool streaming-io

Intermediate 8 steps

package validate
 
import (
	"fmt"

Struct validation with reflection in Go

reflection struct-tags validation

Intermediate 7 steps

ruby

class ParamCoercer
  TYPES = {
    integer: ->(v) { Integer(v) },
    float:   ->(v) { Float(v) },

Coercing request params by schema in Ruby

lambdas type-coercion lookup-table

Intermediate 7 steps

python

from datetime import datetime
from zoneinfo import ZoneInfo

Timezone-safe datetime handling in Python

timezones datetime normalization

Intermediate 4 steps

rust

use std::env;
use std::time::Duration;
 
#[derive(Debug, Clone)]

Loading typed config from environment variables in Rust

configuration error-handling generics

Intermediate 7 steps

javascript

async function copyToClipboard(text) {
  if (navigator.clipboard && window.isSecureContext) {
    try {
      await navigator.clipboard.writeText(text);