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

javascript

import { Suspense } from 'react';
import { searchProducts } from '@/lib/products';
import SearchInput from './search-input';

Streaming search results in a Next.js Server Component

server-components suspense streaming

Intermediate 8 steps

func (h *ExportHandler) StreamExport(c *gin.Context) {
	datasetID := c.Param("id")
 
	ctx := c.Request.Context()

Streaming NDJSON progress with Gin

streaming goroutines channels

Advanced 8 steps

python

from functools import wraps
 
from flask import Blueprint, abort, jsonify
from flask_login import current_user, login_required

Building an admin-only decorator in Flask

decorators authorization access-control

Intermediate 7 steps

python

from django.urls import path, include
 
app_name = "api"

How URL-namespaced API versioning works in Django

api versioning url routing namespaces

Intermediate 8 steps

javascript

import { notFound } from 'next/navigation'
import { Suspense } from 'react'
import { getPostBySlug, getRelatedPosts } from '@/lib/posts'
import { RelatedPosts } from '@/components/related-posts'

Building a dynamic blog post page in Next.js

server-components dynamic-routing metadata

Intermediate 8 steps

type CreateUserInput struct {
	Name     string `json:"name" binding:"required,min=2,max=64"`
	Email    string `json:"email" binding:"required,email"`
	Password string `json:"password" binding:"required,min=8"`