FastAPI File Uploads and Form Data

from fastapi import FastAPI, UploadFile, File, HTTPException, status
import shutil
from pathlib import Path

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

@app.post('/upload')
async def upload_document(file: UploadFile = File(...)):
    # 1. MIME Type Validation (Don't trust the extension!)
    if file.content_type not in ['application/pdf', 'image/jpeg']:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail="Invalid file type. Only PDF and JPEG are allowed."
        )

    # 2. Size Validation using file object metadata
    # Note: .size is available in newer FastAPI/Starlette versions
    MAX_SIZE = 10 * 1024 * 1024  # 10MB
    real_file_size = 0
    
    # Stream content to disk to avoid memory spikes
    save_path = UPLOAD_DIR / file.filename
    with open(save_path, "wb") as buffer:
        while chunk := await file.read(1024 * 1024):  # Read in 1MB chunks
            real_file_size += len(chunk)
            if real_file_size > MAX_SIZE:
                raise HTTPException(status_code=413, detail="File too large")
            buffer.write(chunk)

    return {
        'filename': file.filename,
        'saved_at': str(save_path),
        'final_size': real_file_size
    }

from fastapi import FastAPI, UploadFile, File, Form, status
from typing import Annotated

app = FastAPI()

@app.post('/forge/profile-update', status_code=status.HTTP_201_CREATED)
async def update_profile(
    username: Annotated[str, Form(...)],
    bio: Annotated[str, Form(min_length=10)],
    # Optional file: defaults to None if not in the request
    avatar: UploadFile | None = File(None) 
):
    response = {
        "user_id": 1024, 
        "username": username, 
        "bio": bio,
        "avatar_received": False
    }
    
    if avatar:
        # Process avatar (e.g., upload to S3 or resize)
        response["avatar_received"] = True
        response["avatar_name"] = avatar.filename
        
    return response

from fastapi import FastAPI, UploadFile, File, HTTPException

app = FastAPI()
MAX_FILES = 10
MAX_TOTAL_SIZE = 100 * 1024 * 1024  # 100MB

@app.post('/upload-multiple')
async def upload_multiple(files: list[UploadFile] = File(...)):
    if len(files) > MAX_FILES:
        raise HTTPException(400, detail=f"Maximum {MAX_FILES} files allowed")
    
    total_size = 0
    saved_files = []
    
    for file in files:
        # Stream file contents to disk
        content = b''
        while chunk := await file.read(1024 * 1024):
            content += chunk
            total_size += len(chunk)
            if total_size > MAX_TOTAL_SIZE:
                raise HTTPException(413, detail="Total upload size exceeds limit")
        
        with open(f"uploads/{file.filename}", "wb") as f:
            f.write(content)
        saved_files.append(file.filename)
    
    return {"saved_files": saved_files, "total_size": total_size}

import magic
from fastapi import UploadFile, HTTPException

ALLOWED_MIME_TYPES = {
    'image/jpeg', 'image/png', 'application/pdf'
}

async def validate_file_type(file: UploadFile):
    # Read first 2048 bytes for magic detection
    chunk = await file.read(2048)
    await file.seek(0)  # Reset stream position
    
    mime = magic.from_buffer(chunk, mime=True)
    if mime not in ALLOWED_MIME_TYPES:
        raise HTTPException(400, detail=f"File type {mime} is not allowed. Only {ALLOWED_MIME_TYPES}")
    return True

import re
import uuid
from pathlib import Path

def sanitize_filename(original: str) -> str:
    # Remove directory separators and replace spaces
    safe = re.sub(r'[\\/]', '', original)
    safe = re.sub(r'[^a-zA-Z0-9._-]', '_', safe)
    # Truncate to avoid filesystem limits (255 chars typical)
    return safe[:255]

def generate_unique_filename(original: str) -> str:
    ext = Path(original).suffix
    return f"{uuid.uuid4().hex}{ext}"

# Usage:
# original = "../../etc/passwd"
# safe = sanitize_filename(original)  -> "..__etc_passwd" (after replacements)
# unique = generate_unique_filename(original) -> "a1b2c3d4..."