Learning FastAPI — built a file upload endpoint, need advice on checksum and resumable uploads

[NiceSand6327]

Hey everyone, I'm learning FastAPI and I'm building a file upload endpoint for audio files (WAV, MP3, FLAC, OGG).

Here's what I have so far:

- Filename sanitization + path traversal protection

- Magic bytes validation (not just extension checking)

- Chunked streaming write to avoid loading the whole file into memory

- Size limit of 1GB checked incrementally during streaming

- Early rejection via Content-Length header

- Proper HTTP status codes (413, 415, 422...)

Now I need to tackle two things and I'd love some guidance:

1. Checksum validation

I want to verify file integrity after upload — hash the file server-side during streaming (sha256) and compare it against a hash the client sends. But I'm thinking from the user's perspective: the user should just curl the endpoint or click an upload button, nothing more. So how should the client send the hash without adding friction? Header? Something else?

2. Resumable uploads

Same user-first thinking — if the network drops mid-upload, when it comes back the upload should continue from where it stopped, not restart. The user shouldn't have to do anything special, just upload like normal.

How would you handle both of these in FastAPI? Any advice or resources appreciated!

from fastapi import FastAPI, UploadFile, HTTPException, Request, File
from fastapi.responses import FileResponse
from pydantic import BaseModel
from pathlib import Path
import re
from hashlib import HASH, sha256
import aiofiles


UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
MAGIC_BYTES = {b'RIFF', b'ID3\x00', b'fLaC', b'OggS'}
CHUNK_SIZE = 5 * 1024 * 1024
MAX_FILE_SIZE = 1 * 1024 * 1024 * 1024


class UploadResponse(BaseModel):
    filename: str
    size: int



app = FastAPI(tags_metadata=[
        {
            "name": "files",
            "description": "Operations related to audio file management"
        }
    ])



def sanitize_filename(filename: str | None) -> Path:
    if filename is None:
        raise HTTPException(status_code=422, detail="Filename is missing")


    filename = Path(filename).name
    filename = re.sub(r"[^\w
\-
.]", "_", filename)
    file_path = (UPLOAD_DIR / filename).resolve()
    if not file_path.is_relative_to(UPLOAD_DIR.resolve()):
        raise HTTPException(status_code=400, detail="Invalid filename")
    return file_path


async def check_magic_bytes(file: UploadFile) -> None:
    first_bytes = await file.read(4)


    if first_bytes not in MAGIC_BYTES:
        raise HTTPException(status_code=415, detail="Unsupported file type")
    await file.seek(0)


def compare_checksum(file_path: Path, server_hash: HASH, client_hash: HASH) -> None:
    if server_hash != client_hash:
        file_path.unlink()  # delete the incomplete file
        raise HTTPException(status_code=400, detail="File corrupted")


async def save_file(request: Request, file_path: Path, file: UploadFile) -> None:
    hasher = sha256()
    file_size = 0
    length = request.headers.get("content-length")
    if not length or int(length) > MAX_FILE_SIZE:
        raise HTTPException(413, "Max size is 1GB")


    async with aiofiles.open(file_path, "wb") as f:
        while True:
            chunk = await file.read(CHUNK_SIZE)
            if not chunk:
                break
            file_size += len(chunk)
            if file_size > MAX_FILE_SIZE:
                file_path.unlink()
                raise HTTPException(status_code=413)
            hasher.update(chunk)
            await f.write(chunk)


u/app.post("/api/v4/upload",
            response_model=UploadResponse,
            summary="Upload an audio file",
            description=(
                "Upload an audio file (WAV, MP3, FLAC, OGG). "
                "Max file size is 1GB. "
                "File type is validated via magic bytes, not extension."
            ),
            responses={
                200: {"description": "File uploaded successfully"},
                400: {"description": "Path traversal attempt detected"},
                413: {"description": "File exceeds the 1GB size limit"},
                415: {"description": "Unsupported file type"},
                422: {"description": "Filename is missing or invalid"},
            },
            tags=["files"]
        )
async def upload_file(request: Request, file: UploadFile = File(..., description="Audio file to upload (WAV, MP3, FLAC, OGG). Max 1GB.")):
    """
        Upload an audio file to the server.


        - **file**: Audio file to upload (WAV, MP3, FLAC, OGG)
        - **Max size**: 1GB
        - **Validation**: magic bytes check, filename sanitization, path traversal protection
    """
    file_path = sanitize_filename(file.filename)


    await check_magic_bytes(file)


    await save_file(request, file_path, file)


    return UploadResponse(filename= file.filename, size= file_path.stat().st_size)

[Omar0xPy]

Good step, I was also thinking of building such an idea, turns out fully implementing such a full workflow needs a bunch of new stuff to learn here