File Upload & Download Patterns

Master multipart uploads, streaming for large files, S3 direct uploads with presigned URLs, comprehensive file validation, virus scanning integration, resumable chunked uploads, and production-ready download patterns with proper headers.

Why File Handling Matters

File uploads and downloads are fundamental to modern APIs, profile pictures, document storage, data exports, media uploads. But file handling is also one of the biggest security and performance challenges. A single poorly validated file upload can compromise your entire system.Loading a 500MB file into memory crashes your server. Serving files without proper headers enables XSS attacks.

Common Security Risks

Malware/virus uploads (ransomware, trojans)
Path traversal attacks (../../etc/passwd)
Unrestricted file types (executable uploads)
DoS via large files (memory exhaustion)
MIME type spoofing (bypass validation)
XSS via served files (SVG, HTML uploads)

What You'll Learn

Multipart/form-data protocol and FastAPI handling
Streaming uploads (avoid loading into memory)
S3 direct uploads with presigned URLs
Multi-layer validation (extension, MIME, magic bytes)
Virus scanning integration (ClamAV)
Chunked/resumable uploads for large files
Download patterns (Content-Type, Content-Disposition)
Progress tracking and temporary file cleanup

This lesson builds on Lesson 3 (basic multipart file uploads and streaming) and integrates with Lesson 5 (security validation), Lesson 20 (database metadata storage), and Lesson 21(background jobs for file processing). You'll implement production patterns used by Dropbox, Google Drive, and AWS S3, going far beyond basic uploads to handle security, scalability, and reliability at scale.

File Upload Fundamentals

File uploads use the multipart/form-data content type, which encodes binary file data alongside form fields. Lesson 3 introduced basic file upload handling with FastAPI's UploadFile. Now we'll dive deeper into production patterns: multi-layer validation, security scanning, and scaling to millions of uploads.

The Multipart/Form-Data Protocol

When a client uploads a file, the HTTP request contains boundaries separating different parts (fields and files). FastAPI's UploadFile handles parsing automatically.

# Basic file upload with FastAPI
from fastapi import FastAPI, UploadFile, File, HTTPException
from pathlib import Path
import shutil

app = FastAPI()

# Storage configuration
UPLOAD_DIR = Path("/tmp/uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".pdf", ".doc", ".docx"}

@app.post("/api/upload")
async def upload_file(file: UploadFile = File(...)):
    """
    Basic file upload with size and type validation.
    Remember Lesson 5: Never trust user input!
    """
    # Validate file extension
    file_ext = Path(file.filename).suffix.lower()
    if file_ext not in ALLOWED_EXTENSIONS:
        raise HTTPException(
            status_code=415,  # Unsupported Media Type (Lesson 6)
            detail=f"File type {file_ext} not allowed. Allowed: {ALLOWED_EXTENSIONS}"
        )

    # Check file size (read in chunks to avoid memory issues)
    file_size = 0
    await file.seek(0)  # Reset to start
    chunk_size = 1024 * 1024  # 1MB chunks

    # Validate size by reading header first
    first_chunk = await file.read(chunk_size)
    file_size += len(first_chunk)

    if file_size > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,  # Payload Too Large
            detail=f"File too large. Max size: {MAX_FILE_SIZE / 1024 / 1024}MB"
        )

    # Generate safe filename (prevent path traversal)
    safe_filename = f"{uuid.uuid4()}{file_ext}"
    file_path = UPLOAD_DIR / safe_filename

    # Save file
    with file_path.open("wb") as buffer:
        buffer.write(first_chunk)
        while chunk := await file.read(chunk_size):
            file_size += len(chunk)
            if file_size > MAX_FILE_SIZE:
                file_path.unlink()  # Delete partial file
                raise HTTPException(status_code=413, detail="File too large")
            buffer.write(chunk)

    return {
        "filename": file.filename,
        "stored_as": safe_filename,
        "size": file_size,
        "content_type": file.content_type,
        "url": f"/api/files/{safe_filename}"
    }

Security Warning:Never use user-provided filenames directly. They can contain path traversal attempts like ../../etc/passwd. Always generate unique, sanitized filenames.

Bad vs Good File Handling

❌ Bad: Loading Entire File into Memory

# DON'T DO THIS - OOM on large files
content = await file.read()  # Loads entire file
# 500MB file = 500MB RAM used!

✅ Good: Streaming in Chunks

# Stream in 1MB chunks
while chunk := await file.read(1024*1024):
    process_chunk(chunk)
# Only 1MB RAM used at a time

❌ Bad: Trusting Client MIME Type

# Client says it's an image...
if file.content_type == "image/png":
    save(file)  # But it's actually malware!

✅ Good: Validate with Magic Bytes

# Check actual file signature
import magic
actual_type = magic.from_buffer(chunk)
if "PNG" not in actual_type: reject()

Handling Large Files - Streaming

Loading large files into memory causes crashes. A 500MB upload uses 500MB of RAM, multiply that by 100 concurrent uploads and your server runs out of memory. Always stream files in chunks.

Memory Usage: Loading vs Streaming

┌─────────────────────────────────┐
│ Loading Entire File (500MB)     │
└─────────────────────────────────┘
RAM: ████████████████████████████ 500MB used

┌─────────────────────────────────┐
│ Streaming in 1MB Chunks         │
└─────────────────────────────────┘
RAM: █ 1MB used (500x less memory!)

Rule of thumb: Stream any file larger than 10MB. For files over 100MB, consider chunked uploads with resume capability.

Streaming Upload Implementation

# Streaming file upload to avoid memory issues
import aiofiles
from fastapi import UploadFile, BackgroundTasks
from typing import AsyncGenerator

CHUNK_SIZE = 1024 * 1024  # 1MB chunks

@app.post("/api/upload/stream")
async def upload_file_streaming(
    file: UploadFile = File(...),
    background_tasks: BackgroundTasks = None
):
    """
    Stream large file uploads without loading into memory.
    Suitable for files up to 1GB.
    """
    file_path = UPLOAD_DIR / f"{uuid.uuid4()}{Path(file.filename).suffix}"
    file_size = 0

    try:
        # Stream to disk in chunks
        async with aiofiles.open(file_path, "wb") as f:
            while chunk := await file.read(CHUNK_SIZE):
                file_size += len(chunk)

                # Enforce size limit
                if file_size > MAX_FILE_SIZE:
                    file_path.unlink()  # Delete partial file
                    raise HTTPException(
                        status_code=413,
                        detail=f"File exceeds {MAX_FILE_SIZE / 1024 / 1024}MB limit"
                    )

                await f.write(chunk)

        # Process file in background (virus scan, thumbnail generation, etc.)
        # Reference Lesson 21 - Background Jobs
        if background_tasks:
            background_tasks.add_task(process_uploaded_file, file_path)

        return {
            "message": "File uploaded successfully",
            "size": file_size,
            "path": str(file_path),
            "processing": "background" if background_tasks else "complete"
        }

    except Exception as e:
        # Clean up on failure
        if file_path.exists():
            file_path.unlink()
        raise

async def process_uploaded_file(file_path: Path):
    """Background task to process uploaded file"""
    # Virus scan (see Section 5)
    scan_result = await scan_for_viruses(file_path)
    if scan_result["infected"]:
        file_path.unlink()
        logger.error(f"Infected file deleted: {file_path}")
        return

    # Generate thumbnail for images, etc.
    # Store metadata in database (Lesson 20)
    save_file_metadata_to_db(file_path, scan_result)

Performance Tip:Use aiofiles for async file I/O. This prevents blocking the event loop during disk writes. For very large files (>1GB), consider chunked uploads (next section).

Chunked Uploads with Resume Capability

For files over 100MB, implement chunked uploads. The client splits the file into chunks (e.g., 5MB each) and uploads them sequentially. If the connection drops, resume from the last successful chunk.

# Chunked upload with resume capability
from pydantic import BaseModel
from typing import Optional

class ChunkMetadata(BaseModel):
    upload_id: str
    chunk_number: int
    total_chunks: int
    filename: str

# In-memory store for upload sessions (use Redis in production, Lesson 8)
upload_sessions = {}

@app.post("/api/upload/chunk")
async def upload_chunk(
    file: UploadFile = File(...),
    upload_id: str = Form(...),
    chunk_number: int = Form(...),
    total_chunks: int = Form(...),
    original_filename: str = Form(...)
):
    """
    Upload a single chunk of a large file.
    Client sends chunks sequentially, can resume on failure.
    """
    # Initialize upload session
    if upload_id not in upload_sessions:
        safe_filename = f"{uuid.uuid4()}_{Path(original_filename).suffix}"
        upload_sessions[upload_id] = {
            "filename": safe_filename,
            "total_chunks": total_chunks,
            "received_chunks": set(),
            "file_path": UPLOAD_DIR / safe_filename
        }

    session = upload_sessions[upload_id]

    # Skip if chunk already received (resume scenario)
    if chunk_number in session["received_chunks"]:
        return {
            "message": f"Chunk {chunk_number} already received",
            "progress": len(session["received_chunks"]) / total_chunks * 100
        }

    # Append chunk to file
    chunk_data = await file.read()
    mode = "ab" if session["file_path"].exists() else "wb"

    async with aiofiles.open(session["file_path"], mode) as f:
        await f.write(chunk_data)

    session["received_chunks"].add(chunk_number)

    # Check if upload complete
    if len(session["received_chunks"]) == total_chunks:
        file_path = session["file_path"]
        del upload_sessions[upload_id]  # Clean up session

        return {
            "message": "Upload complete",
            "filename": session["filename"],
            "url": f"/api/files/{session['filename']}",
            "progress": 100
        }

    return {
        "message": f"Chunk {chunk_number}/{total_chunks} received",
        "progress": len(session["received_chunks"]) / total_chunks * 100
    }

@app.get("/api/upload/{upload_id}/status")
async def get_upload_status(upload_id: str):
    """Check upload progress (for resume)"""
    if upload_id not in upload_sessions:
        raise HTTPException(status_code=404, detail="Upload session not found")

    session = upload_sessions[upload_id]
    return {
        "upload_id": upload_id,
        "received_chunks": sorted(session["received_chunks"]),
        "total_chunks": session["total_chunks"],
        "progress": len(session["received_chunks"]) / session["total_chunks"] * 100
    }

Production Note:Store upload sessions in Redis (Lesson 8), not in-memory. In-memory storage is lost on server restart. Set Redis TTL to 24 hours to automatically clean up abandoned uploads.

Cloud Storage - S3 Direct Uploads

Uploading files through your server wastes bandwidth and CPU. Direct-to-S3 uploadslet clients upload files directly to S3 using presigned URLs, bypassing your server entirely. This is how Dropbox, Google Drive, and other file services scale to billions of files.

Server Upload vs Direct S3 Upload

Aspect	Server Upload	Direct S3 Upload
Bandwidth	2x (client → server → S3)	1x (client → S3 directly)
Server Load	High (processes all files)	Low (only generates URLs)
Speed	Slower (double network hop)	Faster (direct to S3)
Scaling	Need more servers	S3 scales automatically
Cost	Server bandwidth + S3	S3 only (cheaper)
Validation	Easy (before upload)	Limited (via S3 policies)

Presigned URLs for Secure Uploads

A presigned URL is a temporary, signed URL that grants upload permissions without exposing AWS credentials. URLs expire after a set time (typically 15 minutes).

# Generate presigned S3 upload URL
import boto3
from botocore.config import Config
from datetime import datetime, timedelta, timezone

# Configure S3 client (Lesson 15 - Deployment)
s3_client = boto3.client(
    's3',
    region_name='us-east-1',
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
    config=Config(signature_version='s3v4')
)

BUCKET_NAME = "my-app-uploads"

@app.post("/api/upload/presigned-url")
async def generate_upload_url(
    filename: str,
    content_type: str,
    file_size: int
):
    """
    Generate presigned S3 URL for direct browser upload.
    Client uploads to this URL without server involvement.
    """
    # Validate before generating URL
    if file_size > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail=f"File too large. Max: {MAX_FILE_SIZE / 1024 / 1024}MB"
        )

    allowed_types = ["image/jpeg", "image/png", "application/pdf"]
    if content_type not in allowed_types:
        raise HTTPException(
            status_code=415,
            detail=f"Content type {content_type} not allowed"
        )

    # Generate unique S3 key
    file_ext = mimetypes.guess_extension(content_type) or ""
    s3_key = f"uploads/{datetime.now(timezone.utc).strftime('%Y/%m/%d')}/{uuid.uuid4()}{file_ext}"

    # Generate presigned URL (expires in 15 minutes)
    presigned_url = s3_client.generate_presigned_url(
        'put_object',
        Params={
            'Bucket': BUCKET_NAME,
            'Key': s3_key,
            'ContentType': content_type,
            'ContentLength': file_size,
        },
        ExpiresIn=900,  # 15 minutes
        HttpMethod='PUT'
    )

    return {
        "upload_url": presigned_url,
        "s3_key": s3_key,
        "expires_at": (datetime.now(timezone.utc) + timedelta(minutes=15)).isoformat(),
        "method": "PUT",
        "headers": {
            "Content-Type": content_type,
            "Content-Length": str(file_size)
        }
    }

Client-Side S3 Upload (JavaScript)

// Client-side: Upload file directly to S3
async function uploadFileToS3(file) {
    try {
        // Step 1: Get presigned URL from backend
        const response = await fetch('/api/upload/presigned-url', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                filename: file.name,
                content_type: file.type,
                file_size: file.size
            })
        });

        if (!response.ok) {
            throw new Error(`Failed to get upload URL: ${response.status}`);
        }

        const { upload_url, s3_key, headers } = await response.json();

        // Step 2: Upload directly to S3 (bypasses backend!)
        const uploadResponse = await fetch(upload_url, {
            method: 'PUT',
            headers: {
                'Content-Type': headers['Content-Type'],
                'Content-Length': headers['Content-Length']
            },
            body: file  // Binary file data
        });

        if (!uploadResponse.ok) {
            throw new Error(`S3 upload failed: ${uploadResponse.status}`);
        }

        // Step 3: Notify backend that upload completed
        await fetch('/api/upload/complete', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                s3_key: s3_key,
                filename: file.name,
                size: file.size,
                content_type: file.type
            })
        });

        console.log('File uploaded successfully to S3!');
        return { s3_key, success: true };

    } catch (error) {
        console.error('Upload failed:', error);
        throw error;
    }
}

Callback After S3 Upload

After the client uploads to S3, they notify your server. You then verify the upload, store metadata in the database (Lesson 20), and trigger background processing (Lesson 21).

# Callback endpoint after S3 upload
from pydantic import BaseModel

class UploadComplete(BaseModel):
    s3_key: str
    filename: str
    size: int
    content_type: str

@app.post("/api/upload/complete")
async def upload_complete(
    data: UploadComplete,
    background_tasks: BackgroundTasks,
    current_user: User = Depends(get_current_user)
):
    """
    Called by client after successful S3 upload.
    Verifies upload, stores metadata, triggers processing.
    """
    # Verify file actually exists in S3
    try:
        s3_client.head_object(Bucket=BUCKET_NAME, Key=data.s3_key)
    except:
        raise HTTPException(
            status_code=400,
            detail="File not found in S3. Upload may have failed."
        )

    # Store metadata in database (Lesson 20 - connection pooling)
    file_record = {
        "id": str(uuid.uuid4()),
        "user_id": current_user.id,
        "s3_key": data.s3_key,
        "filename": data.filename,
        "size": data.size,
        "content_type": data.content_type,
        "uploaded_at": datetime.now(timezone.utc),
        "status": "processing"  # Will be updated after scan
    }

    await db.execute(
        "INSERT INTO files (...) VALUES (...)",
        file_record
    )

    # Queue background processing (Lesson 21 - Celery)
    # - Virus scan (see Section 5)
    # - Generate thumbnail for images
    # - Extract metadata
    background_tasks.add_task(
        process_s3_file,
        s3_key=data.s3_key,
        file_id=file_record["id"]
    )

    return {
        "message": "Upload registered successfully",
        "file_id": file_record["id"],
        "status": "processing",
        "url": f"/api/files/{file_record['id']}"
    }

Best Practice:Always verify the S3 upload succeeded before storing metadata. Use head_object()to check existence. This prevents database entries for files that don't exist.

File Downloads & Streaming

Serving files requires proper HTTP headers. Content-Type tells the browser what kind of file it is. Content-Disposition controls whether it's displayed inline or downloaded. For large files, use streaming to avoid loading into memory.

File Download with Proper Headers

# File download endpoint with proper headers
from fastapi.responses import FileResponse, StreamingResponse
import mimetypes

@app.get("/api/files/{file_id}/download")
async def download_file(
    file_id: str,
    inline: bool = False,  # inline=True displays in browser, False downloads
    current_user: User = Depends(get_current_user)
):
    """
    Download file with proper Content-Type and Content-Disposition.
    Supports inline display (PDFs, images) or forced download.
    """
    # Fetch file metadata from database
    file_record = await db.fetch_one(
        "SELECT * FROM files WHERE id = :id AND user_id = :user_id",
        {"id": file_id, "user_id": current_user.id}
    )

    if not file_record:
        raise HTTPException(status_code=404, detail="File not found")

    # Determine MIME type
    content_type = file_record["content_type"]
    if not content_type:
        content_type, _ = mimetypes.guess_type(file_record["filename"])
        content_type = content_type or "application/octet-stream"

    # Set Content-Disposition header
    disposition_type = "inline" if inline else "attachment"
    headers = {
        "Content-Disposition": f'{disposition_type}; filename="{file_record["filename"]}"',
        "Content-Type": content_type,
        "Content-Length": str(file_record["size"])
    }

    # For local files
    if file_record["storage"] == "local":
        file_path = Path(file_record["path"])
        return FileResponse(
            file_path,
            headers=headers,
            media_type=content_type
        )

    # For S3 files, generate presigned download URL
    elif file_record["storage"] == "s3":
        download_url = s3_client.generate_presigned_url(
            'get_object',
            Params={
                'Bucket': BUCKET_NAME,
                'Key': file_record["s3_key"],
                'ResponseContentDisposition': headers["Content-Disposition"],
                'ResponseContentType': content_type
            },
            ExpiresIn=3600  # 1 hour
        )

        # Redirect to S3 (avoids proxying through server)
        return {"download_url": download_url}

Streaming Large File Downloads

For files larger than 10MB, stream the response instead of loading the entire file into memory. This prevents OOM errors and allows concurrent downloads.

# Stream large file downloads
from typing import AsyncIterator

async def file_iterator(file_path: Path, chunk_size: int = 1024 * 1024) -> AsyncIterator[bytes]:
    """
    Async generator that yields file chunks.
    Only loads chunk_size bytes into memory at a time.
    """
    async with aiofiles.open(file_path, "rb") as f:
        while chunk := await f.read(chunk_size):
            yield chunk

@app.get("/api/files/{file_id}/stream")
async def stream_file_download(
    file_id: str,
    current_user: User = Depends(get_current_user)
):
    """
    Stream large file download.
    Suitable for files > 10MB to avoid memory issues.
    """
    file_record = await get_file_record(file_id, current_user.id)
    file_path = Path(file_record["path"])

    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    headers = {
        "Content-Disposition": f'attachment; filename="{file_record["filename"]}"',
        "Content-Type": file_record["content_type"],
        "Content-Length": str(file_record["size"]),
        "Accept-Ranges": "bytes"  # Enable range requests (partial downloads)
    }

    return StreamingResponse(
        file_iterator(file_path),
        headers=headers,
        media_type=file_record["content_type"]
    )

Range Requests (Partial Downloads)

Range requests allow clients to download parts of a file, useful for video streaming, resuming interrupted downloads, and seeking in media files.

# Support Range requests for partial downloads
from fastapi import Request

@app.get("/api/files/{file_id}")
async def serve_file_with_ranges(
    file_id: str,
    request: Request,
    current_user: User = Depends(get_current_user)
):
    """
    Serve file with Range request support.
    Enables video streaming, resumable downloads, seeking.
    """
    file_record = await get_file_record(file_id, current_user.id)
    file_path = Path(file_record["path"])
    file_size = file_record["size"]

    # Check if client requested a range
    range_header = request.headers.get("Range")

    if not range_header:
        # No range requested, send entire file
        return StreamingResponse(
            file_iterator(file_path),
            headers={"Content-Type": file_record["content_type"]},
            media_type=file_record["content_type"]
        )

    # Parse Range header (e.g., "bytes=0-1023")
    range_match = re.match(r"bytes=(d+)-(d*)", range_header)
    if not range_match:
        raise HTTPException(status_code=400, detail="Invalid Range header")

    start = int(range_match.group(1))
    end = int(range_match.group(2)) if range_match.group(2) else file_size - 1

    if start >= file_size or end >= file_size:
        raise HTTPException(
            status_code=416,  # Range Not Satisfiable
            detail=f"Range out of bounds. File size: {file_size}"
        )

    # Read and return requested range
    chunk_size = end - start + 1

    async def range_iterator():
        async with aiofiles.open(file_path, "rb") as f:
            await f.seek(start)
            remaining = chunk_size
            while remaining > 0:
                read_size = min(1024 * 1024, remaining)  # 1MB chunks
                chunk = await f.read(read_size)
                if not chunk:
                    break
                remaining -= len(chunk)
                yield chunk

    headers = {
        "Content-Range": f"bytes {start}-{end}/{file_size}",
        "Content-Length": str(chunk_size),
        "Content-Type": file_record["content_type"],
        "Accept-Ranges": "bytes"
    }

    return StreamingResponse(
        range_iterator(),
        status_code=206,  # Partial Content
        headers=headers,
        media_type=file_record["content_type"]
    )

Use Case:Range requests are essential for video streaming. When a user seeks to 2:30 in a video, the player sends Range: bytes=25000000- to start from that position instead of downloading the entire file.

Security - Validation & Virus Scanning

File uploads are one of the biggest security risks. A single malicious file can compromise your entire system. Never trust user input. Validate at multiple layers and always scan for malware. Remember Lesson 5, defense in depth.

Multi-Layer File Validation

Extension Check

First layer: check file extension (.jpg, .pdf, etc.)

Weakness: Easily bypassed (malware.exe → malware.jpg)

MIME Type Check

Second layer: check Content-Type header from client

Weakness: Client-controlled, can be spoofed

Magic Bytes Check

Third layer: check file signature (first bytes of file)

Reliable: Cannot be spoofed easily, validates actual content

# Complete multi-layer file validation pipeline
import magic  # python-magic library
import hashlib

ALLOWED_MIME_TYPES = {
    "image/jpeg": [".jpg", ".jpeg"],
    "image/png": [".png"],
    "application/pdf": [".pdf"],
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": [".docx"]
}

# Magic bytes signatures for common file types
MAGIC_BYTES = {
    "image/jpeg": [b"ÿØÿ"],
    "image/png": [b"PNG

"],
    "application/pdf": [b"%PDF-"],
    "application/zip": [b"PK", b"PK"]  # ZIP/DOCX
}

async def validate_file(file: UploadFile) -> dict:
    """
    Comprehensive file validation with multiple layers.
    Returns validation result with details.
    """
    result = {
        "valid": True,
        "errors": [],
        "file_type": None,
        "file_hash": None
    }

    # Layer 1: Extension validation
    file_ext = Path(file.filename).suffix.lower()
    allowed_extensions = [ext for exts in ALLOWED_MIME_TYPES.values() for ext in exts]

    if file_ext not in allowed_extensions:
        result["valid"] = False
        result["errors"].append(f"Extension {file_ext} not allowed")
        return result

    # Layer 2: Read first chunk for further validation
    first_chunk = await file.read(8192)  # Read 8KB
    await file.seek(0)  # Reset for later reading

    # Layer 3: Magic bytes validation (file signature)
    detected_type = None
    for mime_type, signatures in MAGIC_BYTES.items():
        for signature in signatures:
            if first_chunk.startswith(signature):
                detected_type = mime_type
                break
        if detected_type:
            break

    if not detected_type:
        # Use python-magic for more comprehensive detection
        detected_type = magic.from_buffer(first_chunk, mime=True)

    # Verify detected type matches allowed types
    if detected_type not in ALLOWED_MIME_TYPES:
        result["valid"] = False
        result["errors"].append(
            f"File signature indicates {detected_type}, which is not allowed"
        )
        return result

    # Verify extension matches detected type
    if file_ext not in ALLOWED_MIME_TYPES[detected_type]:
        result["valid"] = False
        result["errors"].append(
            f"Extension {file_ext} doesn't match file type {detected_type}"
        )
        return result

    # Layer 4: Calculate file hash (for deduplication and integrity)
    hasher = hashlib.sha256()
    hasher.update(first_chunk)

    while chunk := await file.read(1024 * 1024):  # 1MB chunks
        hasher.update(chunk)

    await file.seek(0)  # Reset for later use
    result["file_hash"] = hasher.hexdigest()
    result["file_type"] = detected_type

    return result

@app.post("/api/upload/validated")
async def upload_with_validation(file: UploadFile = File(...)):
    """Upload with comprehensive validation"""
    # Validate file
    validation = await validate_file(file)

    if not validation["valid"]:
        raise HTTPException(
            status_code=415,
            detail={"message": "File validation failed", "errors": validation["errors"]}
        )

    # Check for duplicate (using hash)
    existing = await db.fetch_one(
        "SELECT id FROM files WHERE file_hash = :hash",
        {"hash": validation["file_hash"]}
    )

    if existing:
        return {
            "message": "File already exists (duplicate detected)",
            "file_id": existing["id"]
        }

    # File is valid - proceed with upload
    # ... save file logic ...

    return {"message": "File validated and uploaded", "file_type": validation["file_type"]}

Virus Scanning with ClamAV

Always scan uploaded files for viruses and malware. Integrate with ClamAV (open-source) or commercial services like VirusTotal. Run scans in background jobs (Lesson 21) to avoid blocking the upload.

# Virus scanning integration with ClamAV
import clamd  # pip install clamd

# Connect to ClamAV daemon
clam = clamd.ClamdUnixSocket()  # Or ClamdNetworkSocket() for remote

async def scan_file_for_viruses(file_path: Path) -> dict:
    """
    Scan file for viruses using ClamAV.
    Returns scan result with threat details if found.
    """
    try:
        # Scan file
        scan_result = clam.scan(str(file_path))

        if scan_result is None:
            return {
                "infected": False,
                "clean": True,
                "threat": None
            }

        # Parse result
        file_status = scan_result.get(str(file_path))

        if file_status and file_status[0] == "FOUND":
            return {
                "infected": True,
                "clean": False,
                "threat": file_status[1],  # Threat name (e.g., "Win.Malware.Trojan")
                "action": "quarantine"
            }

        return {"infected": False, "clean": True, "threat": None}

    except clamd.ConnectionError:
        # ClamAV daemon not running
        logger.error("ClamAV daemon not available")
        return {
            "infected": None,
            "clean": False,
            "error": "Antivirus scan unavailable"
        }

@app.post("/api/upload/secure")
async def upload_with_virus_scan(
    file: UploadFile = File(...),
    background_tasks: BackgroundTasks = None
):
    """
    Upload file with virus scanning.
    Scan runs in background to avoid blocking upload.
    """
    # Save file temporarily
    temp_path = UPLOAD_DIR / f"temp_{uuid.uuid4()}{Path(file.filename).suffix}"
    async with aiofiles.open(temp_path, "wb") as f:
        while chunk := await file.read(1024 * 1024):
            await f.write(chunk)

    # Queue virus scan in background (Lesson 21 - Background Jobs)
    if background_tasks:
        background_tasks.add_task(
            scan_and_process_file,
            file_path=temp_path,
            user_id=current_user.id,
            original_filename=file.filename
        )

        return {
            "message": "File uploaded, virus scan in progress",
            "status": "scanning",
            "check_status_at": f"/api/files/{file_id}/status"
        }

    # Or scan synchronously for small files
    scan_result = await scan_file_for_viruses(temp_path)

    if scan_result["infected"]:
        temp_path.unlink()  # Delete infected file
        raise HTTPException(
            status_code=422,
            detail=f"File infected with {scan_result['threat']}. Upload rejected."
        )

    # File is clean - move to permanent storage
    final_path = UPLOAD_DIR / f"{uuid.uuid4()}{Path(file.filename).suffix}"
    temp_path.rename(final_path)

    return {
        "message": "File uploaded and scanned (clean)",
        "scan_result": scan_result,
        "file_path": str(final_path)
    }

async def scan_and_process_file(file_path: Path, user_id: str, original_filename: str):
    """Background task: scan file and update database"""
    scan_result = await scan_file_for_viruses(file_path)

    if scan_result["infected"]:
        # Delete infected file and notify user
        file_path.unlink()
        await notify_user(user_id, f"File {original_filename} infected: {scan_result['threat']}")
        await db.execute(
            "UPDATE files SET status = 'infected', threat = :threat WHERE path = :path",
            {"threat": scan_result["threat"], "path": str(file_path)}
        )
    else:
        # Mark as clean and move to permanent storage
        await db.execute(
            "UPDATE files SET status = 'clean', scanned_at = NOW() WHERE path = :path",
            {"path": str(file_path)}
        )

Critical:Never skip virus scanning for "trusted" users. Attackers often compromise legitimate accounts. Always scan every uploaded file, regardless of source.

Path Traversal Prevention & Filename Sanitization

User-provided filenames can contain path traversal attempts (../../etc/passwd). Always sanitize filenames and use UUIDs for storage.

# Filename sanitization to prevent path traversal
import re
import unicodedata

def sanitize_filename(filename: str, max_length: int = 255) -> str:
    """
    Sanitize user-provided filename to prevent security issues.
    - Removes path components (../../)
    - Removes special characters
    - Normalizes unicode
    - Limits length
    """
    # Normalize unicode (NFD = decomposed form)
    filename = unicodedata.normalize('NFD', filename)

    # Remove non-ASCII characters
    filename = filename.encode('ascii', 'ignore').decode('ascii')

    # Remove path separators
    filename = filename.replace('/', '').replace('\\', '')

    # Remove dangerous characters
    filename = re.sub(r'[^ws.-]', '', filename)

    # Remove leading/trailing dots and spaces
    filename = filename.strip('. ')

    # Limit length
    if len(filename) > max_length:
        name, ext = os.path.splitext(filename)
        filename = name[:max_length - len(ext)] + ext

    # Ensure filename isn't empty
    if not filename:
        filename = "unnamed_file"

    return filename

def generate_safe_filename(original_filename: str, use_uuid: bool = True) -> str:
    """
    Generate safe filename for storage.
    Best practice: use UUID to avoid collisions and info disclosure.
    """
    if use_uuid:
        # Recommended: UUID + original extension
        ext = Path(original_filename).suffix.lower()
        return f"{uuid.uuid4()}{ext}"
    else:
        # Alternative: sanitized original name + timestamp
        sanitized = sanitize_filename(Path(original_filename).stem)
        ext = Path(original_filename).suffix.lower()
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        return f"{sanitized}_{timestamp}{ext}"

# Security checklist for file uploads
SECURITY_CHECKLIST = {
    "validate_size": True,          # ✅ Reject files > max size
    "validate_extension": True,     # ✅ Check file extension
    "validate_mime_type": True,     # ✅ Check Content-Type
    "validate_magic_bytes": True,   # ✅ Check file signature
    "scan_for_viruses": True,       # ✅ ClamAV or VirusTotal
    "sanitize_filename": True,      # ✅ Remove path traversal
    "use_uuid_filenames": True,     # ✅ Prevent info disclosure
    "enforce_quotas": True,         # ✅ Per-user storage limits
    "rate_limit_uploads": True,     # ✅ Prevent abuse
    "cleanup_temp_files": True,     # ✅ Delete abandoned uploads
    "separate_upload_domain": True, # ✅ Serve from uploads.example.com (prevent XSS)
    "use_presigned_urls": True,     # ✅ For S3, time-limited access
}

Advanced Patterns

Production file handling requires progress tracking, batch processing, quota management, and automatic cleanup. Integrate with background jobs (Lesson 21) for async processing.

Upload Progress Tracking

For long uploads, clients need progress feedback. Store progress in Redis (Lesson 8) and expose a polling endpoint.

# Upload progress tracking with Redis
import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

@app.post("/api/upload/with-progress")
async def upload_with_progress(
    file: UploadFile = File(...),
    upload_id: str = Form(...)
):
    """Track upload progress in Redis"""
    total_size = 0
    chunk_size = 1024 * 1024  # 1MB

    file_path = UPLOAD_DIR / f"{upload_id}{Path(file.filename).suffix}"

    async with aiofiles.open(file_path, "wb") as f:
        while chunk := await file.read(chunk_size):
            await f.write(chunk)
            total_size += len(chunk)

            # Update progress in Redis
            progress = {
                "upload_id": upload_id,
                "bytes_uploaded": total_size,
                "status": "uploading",
                "updated_at": datetime.now(timezone.utc).isoformat()
            }
            redis_client.setex(
                f"upload_progress:{upload_id}",
                3600,  # Expire after 1 hour
                json.dumps(progress)
            )

    # Mark as complete
    redis_client.setex(
        f"upload_progress:{upload_id}",
        3600,
        json.dumps({
            "upload_id": upload_id,
            "bytes_uploaded": total_size,
            "status": "complete",
            "file_path": str(file_path)
        })
    )

    return {"upload_id": upload_id, "size": total_size, "status": "complete"}

@app.get("/api/upload/{upload_id}/progress")
async def get_upload_progress(upload_id: str):
    """Poll endpoint for upload progress"""
    progress_data = redis_client.get(f"upload_progress:{upload_id}")

    if not progress_data:
        raise HTTPException(status_code=404, detail="Upload not found")

    return json.loads(progress_data)

Batch File Upload with Job Queue

For bulk uploads, queue each file as a separate background job. Use Celery from Lesson 21.

# Batch file processing with Celery (Lesson 21 integration)
from celery_app import celery_app

@celery_app.task(name="process_uploaded_file")
def process_file_task(file_path: str, user_id: str):
    """
    Celery task to process uploaded file:
    - Virus scan
    - Generate thumbnail (images)
    - Extract metadata
    - Store in database
    """
    # Run virus scan
    scan_result = scan_file_for_viruses(Path(file_path))
    if scan_result["infected"]:
        Path(file_path).unlink()
        return {"status": "rejected", "reason": "infected"}

    # Process based on file type
    mime_type = magic.from_file(file_path, mime=True)

    if mime_type.startswith("image/"):
        # Generate thumbnail for images
        generate_thumbnail(file_path)

    elif mime_type == "application/pdf":
        # Extract text from PDF
        extract_pdf_text(file_path)

    # Store metadata in database (Lesson 20)
    save_to_database(file_path, user_id, mime_type)

    return {"status": "complete", "file_path": file_path}

@app.post("/api/upload/batch", status_code=202)
async def upload_batch_files(files: list[UploadFile] = File(...)):
    """
    Upload multiple files, queue for background processing.
    Returns job IDs to check status (202 Accepted pattern from Lesson 6/21).
    """
    job_ids = []

    for file in files:
        # Save file temporarily
        temp_path = UPLOAD_DIR / f"temp_{uuid.uuid4()}{Path(file.filename).suffix}"
        async with aiofiles.open(temp_path, "wb") as f:
            while chunk := await file.read(1024 * 1024):
                await f.write(chunk)

        # Queue processing job
        task = process_file_task.delay(
            file_path=str(temp_path),
            user_id=current_user.id
        )
        job_ids.append(task.id)

    return {
        "message": f"{len(files)} files queued for processing",
        "job_ids": job_ids,
        "check_status_at": "/api/jobs/{job_id}"
    }  # 202 Accepted - processing asynchronously

Automatic Temporary File Cleanup

Clean up old temporary files and abandoned uploads automatically. Schedule with Celery Beat (Lesson 21).

# Scheduled cleanup of temporary files (Celery Beat)
from celery.schedules import crontab
from datetime import datetime, timedelta

@celery_app.task(name="cleanup_temp_files")
def cleanup_temp_files():
    """
    Delete temporary files older than 24 hours.
    Run daily via Celery Beat (Lesson 21).
    """
    temp_dir = Path("/tmp/uploads")
    cutoff_time = datetime.now() - timedelta(hours=24)
    deleted_count = 0

    for file_path in temp_dir.glob("temp_*"):
        try:
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)

            if file_mtime < cutoff_time:
                file_path.unlink()
                deleted_count += 1
                logger.info(f"Deleted temp file: {file_path}")

        except Exception as e:
            logger.error(f"Failed to delete {file_path}: {e}")

    return {
        "deleted_files": deleted_count,
        "cleanup_time": datetime.now().isoformat()
    }

# Configure Celery Beat schedule
celery_app.conf.beat_schedule = {
    "cleanup-temp-files-daily": {
        "task": "cleanup_temp_files",
        "schedule": crontab(hour=3, minute=0),  # 3 AM daily
    }
}

Best Practice:Set up alerts if temp file cleanup fails for multiple days. Growing temp storage indicates abandoned uploads or failing cleanup jobs.

Client-Side Integration

A robust client implementation includes drag-and-drop uploads, progress bars, automatic retries on failure, and resume capability. Here's a production-ready JavaScript client.

Production Upload Client with Retry Logic

// Production-ready file upload client with retry and progress
class FileUploader {
    constructor(apiBaseUrl) {
        this.apiBaseUrl = apiBaseUrl;
        this.maxRetries = 3;
        this.retryDelay = 1000;  // Start with 1s, exponential backoff
    }

    /**
     * Upload file with retry logic and progress tracking
     */
    async uploadFile(file, onProgress = null) {
        let attempts = 0;

        while (attempts < this.maxRetries) {
            try {
                attempts++;
                console.log(`Upload attempt ${attempts}/${this.maxRetries}`);

                const result = await this._attemptUpload(file, onProgress);
                console.log('Upload successful:', result);
                return result;

            } catch (error) {
                console.error(`Upload attempt ${attempts} failed:`, error);

                if (attempts >= this.maxRetries) {
                    throw new Error(`Upload failed after ${this.maxRetries} attempts: ${error.message}`);
                }

                // Exponential backoff
                const delay = this.retryDelay * Math.pow(2, attempts - 1);
                console.log(`Retrying in ${delay}ms...`);
                await this.sleep(delay);
            }
        }
    }

    /**
     * Single upload attempt with progress tracking
     */
    async _attemptUpload(file, onProgress) {
        return new Promise((resolve, reject) => {
            const formData = new FormData();
            formData.append('file', file);

            const xhr = new XMLHttpRequest();

            // Track upload progress
            xhr.upload.addEventListener('progress', (e) => {
                if (e.lengthComputable && onProgress) {
                    const percentComplete = (e.loaded / e.total) * 100;
                    onProgress({
                        loaded: e.loaded,
                        total: e.total,
                        percent: percentComplete.toFixed(2)
                    });
                }
            });

            xhr.addEventListener('load', () => {
                if (xhr.status >= 200 && xhr.status < 300) {
                    resolve(JSON.parse(xhr.responseText));
                } else {
                    reject(new Error(`Upload failed: ${xhr.status} ${xhr.statusText}`));
                }
            });

            xhr.addEventListener('error', () => {
                reject(new Error('Network error during upload'));
            });

            xhr.addEventListener('abort', () => {
                reject(new Error('Upload aborted'));
            });

            xhr.open('POST', `${this.apiBaseUrl}/api/upload`);
            xhr.setRequestHeader('Authorization', `Bearer ${this.getAuthToken()}`);
            xhr.send(formData);
        });
    }

    /**
     * Chunked upload for large files with resume capability
     */
    async uploadLargeFile(file, chunkSize = 5 * 1024 * 1024) {  // 5MB chunks
        const uploadId = this.generateUploadId();
        const totalChunks = Math.ceil(file.size / chunkSize);

        console.log(`Uploading ${file.name} in ${totalChunks} chunks`);

        for (let chunkNumber = 0; chunkNumber < totalChunks; chunkNumber++) {
            const start = chunkNumber * chunkSize;
            const end = Math.min(start + chunkSize, file.size);
            const chunk = file.slice(start, end);

            const formData = new FormData();
            formData.append('file', chunk);
            formData.append('upload_id', uploadId);
            formData.append('chunk_number', chunkNumber);
            formData.append('total_chunks', totalChunks);
            formData.append('original_filename', file.name);

            // Upload chunk with retries
            await this.uploadChunk(formData, chunkNumber, totalChunks);
        }

        return { uploadId, totalChunks, message: 'Upload complete' };
    }

    async uploadChunk(formData, chunkNumber, totalChunks) {
        let attempts = 0;

        while (attempts < this.maxRetries) {
            try {
                attempts++;
                const response = await fetch(`${this.apiBaseUrl}/api/upload/chunk`, {
                    method: 'POST',
                    headers: {
                        'Authorization': `Bearer ${this.getAuthToken()}`
                    },
                    body: formData
                });

                if (!response.ok) {
                    throw new Error(`Chunk upload failed: ${response.status}`);
                }

                const result = await response.json();
                console.log(`Chunk ${chunkNumber + 1}/${totalChunks} uploaded (${result.progress.toFixed(1)}%)`);
                return result;

            } catch (error) {
                if (attempts >= this.maxRetries) {
                    throw error;
                }
                await this.sleep(this.retryDelay * Math.pow(2, attempts - 1));
            }
        }
    }

    sleep(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }

    generateUploadId() {
        return 'upload_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
    }

    getAuthToken() {
        return localStorage.getItem('auth_token');
    }
}

// Usage example
const uploader = new FileUploader('https://api.example.com');

document.getElementById('fileInput').addEventListener('change', async (e) => {
    const file = e.target.files[0];

    if (file.size > 100 * 1024 * 1024) {  // > 100MB
        // Use chunked upload for large files
        await uploader.uploadLargeFile(file);
    } else {
        // Standard upload with progress
        await uploader.uploadFile(file, (progress) => {
            console.log(`Upload progress: ${progress.percent}%`);
            document.getElementById('progressBar').style.width = `${progress.percent}%`;
        });
    }
});

Client Best Practices:

Validate file size/type client-side before upload (faster feedback)
Implement exponential backoff for retries (don't hammer the server)
Show progress bar for uploads >1MB (user feedback)
Use chunked uploads for files >100MB (resume capability)
Handle network errors gracefully (mobile connections drop)

Key Takeaways

1. Never Load Entire Files into Memory

Stream files in chunks (1MB at a time). Loading a 500MB file uses 500MB RAM - multiply by 100 concurrent uploads and you're out of memory. Always usewhile chunk := await file.read(CHUNK_SIZE) pattern.

2. Validate at Multiple Layers (Extension + MIME + Magic Bytes)

Client-provided MIME types and extensions can be spoofed. Always check file signatures (magic bytes) using python-magic. A .jpg file might actually be malware.exe renamed.

3. Direct-to-S3 Uploads Save Bandwidth and Scale Better

Use presigned URLs to let clients upload directly to S3. This eliminates the double network hop (client → server → S3) and reduces server load. Your server only generates URLs, S3 handles storage.

4. Always Scan for Viruses and Malware

Integrate ClamAV or VirusTotal. Run scans in background jobs (Lesson 21) to avoid blocking uploads. Delete infected files immediately. Never skip scanning for "trusted" users, attackers compromise accounts.

5. Use UUIDs for Filenames (Prevent Path Traversal)

Never use user-provided filenames directly. ../../etc/passwd is a classic path traversal attack. Generate unique filenames with UUIDs:f"{uuid.uuid4()}{file_ext}".

6. Implement Upload Quotas and Rate Limiting

Prevent abuse with per-user storage quotas (e.g., 10GB max) and rate limits (e.g., 100 uploads/hour). Use Lesson 8's rate limiting patterns. Monitor for users hitting limits.

7. Content-Disposition Controls Download Behavior

Content-Disposition: attachment forces download.inline displays in browser (PDFs, images). Always set explicitly to control user experience and prevent XSS via HTML/SVG uploads.

8. Chunked Uploads Enable Resume Capability

For files >100MB, implement chunked uploads. Client sends 5MB chunks sequentially. If connection drops, resume from last successful chunk. Store session state in Redis (Lesson 8) with 24h TTL.

9. Clean Up Temporary Files Automatically

Schedule daily cleanup of temp files >24h old using Celery Beat (Lesson 21). Abandoned uploads accumulate quickly. Set alerts if temp storage grows beyond expected levels.

10. Never Serve User Uploads from Main Domain

Serve uploads from separate subdomain (uploads.example.com) or S3/CDN. This prevents XSS attacks via malicious HTML/SVG files. Use Content Security Policy (Lesson 5) to restrict script execution.

Next Steps & Production Checklist

When to Use Each Pattern:

Small files (<10MB): Standard FastAPI UploadFile, stream to disk, validate + virus scan
Medium files (10-100MB): Streaming uploads or direct-to-S3 with presigned URLs
Large files (>100MB): Chunked uploads with resume capability, S3 multipart
Profile pictures: Direct S3 upload, resize in background job (Lesson 21), serve via CDN
Documents: Validate type, virus scan, store metadata in DB (Lesson 20)
Media (video/audio): Direct S3 upload, transcode in background job, use adaptive streaming

Production Security Checklist:

✅ Validate file size (reject > max limit, typically 10-100MB)
✅ Validate extension, MIME type, AND magic bytes (all three layers)
✅ Scan every file for viruses (ClamAV, VirusTotal)
✅ Sanitize filenames, use UUIDs to prevent path traversal
✅ Implement per-user storage quotas (e.g., 10GB)
✅ Rate limit upload endpoints (e.g., 100 uploads/hour)
✅ Use presigned URLs for S3 (time-limited, scoped permissions)
✅ Clean up temp files daily (Celery Beat scheduled task)
✅ Serve uploads from separate domain (prevent XSS)
✅ Set proper Content-Disposition and Content-Type headers

Integration Review: This lesson combines patterns from Lesson 5 (security validation), Lesson 6 (HTTP status codes), Lesson 8 (Redis for progress tracking), Lesson 15 (S3 deployment), Lesson 20 (database metadata storage), and Lesson 21(background jobs for processing). You now have production-ready file handling patterns used by Dropbox, Google Drive, and AWS. Next: apply everything in the Capstone Project!

Back to "Background Jobs & Async Processing"

Web APIsLesson 22

Next: "Pagination Deep Dive"