Post cover image
Bulletproof File Upload Architecture

May 26, 2026

File Uploads in Python APIs — From Broken to Bulletproof

Your API shouldn’t be a middleman between the browser and your storage.

Anas Issath

8 min read

We launched a feature that let users upload profile pictures. The endpoint accepted a file, saved it to disk on the server, and returned the URL. Worked great in development. Worked fine in staging with three test users.

In production, with 2,000 users uploading during the first week, three things happened.

First, a user uploaded a 200MB file — not an image, but a video they'd accidentally selected. Our API server held the entire file in memory while writing it to disk. With four Gunicorn workers and several large uploads happening simultaneously, memory usage spiked to 95% and other requests started failing.

Second, our server's disk filled up. We had 20GB of storage. Profile pictures accumulated without any cleanup. When the disk hit 100%, the database couldn't write its WAL files and the entire application went down.

Third, someone uploaded a PHP file renamed to .jpg. Our Nginx config served it. Nothing exploitable happened because we weren't running PHP, but the security review flagged it as a critical vulnerability — and they were right to.

All three problems have the same root cause: we were treating file uploads as a simple feature when they're actually a distributed systems problem. Here's how to handle them properly.

Why Your API Server Shouldn't Store Files

None

The most common architecture for file uploads — and the one that breaks first — is this:

Browser → Your API → Server Disk

The API receives the file, saves it to the filesystem, and serves it back on request. This has four fundamental problems:

Memory pressure. While your API receives a 50MB file, that data sits in your server's memory. With multiple concurrent uploads, you can exhaust memory and crash other requests.

Single point of failure. Files live on one server's disk. If the server dies, the files are gone. If you scale to multiple servers, files uploaded to server A aren't accessible from server B.

No CDN. Serving files from your API server means every download request competes with API requests for bandwidth, CPU, and connections.

Disk management. You're responsible for disk space monitoring, cleanup, backups, and redundancy. That's infrastructure work you don't need.

The right architecture:

Browser → Your API → Returns presigned URL
Browser → S3 directly (upload)
Browser → CloudFront/CDN (download)

Your API never touches the file data. It generates a presigned URL that gives the browser temporary permission to upload directly to S3. Downloads go through a CDN. Your API stays lightweight, handling only metadata.

The Presigned URL Pattern

None

A presigned URL is a temporary, signed link that grants permission to upload or download a specific file from S3 — without exposing your AWS credentials. The URL contains the permissions, the expiration time, and a cryptographic signature that S3 verifies.

Step 1: Generate Upload URL

# app/services/storage_service.py
import boto3
import uuid
from datetime import datetime
from app.config import settings

s3_client = boto3.client(
    's3',
    aws_access_key_id=settings.AWS_ACCESS_KEY_ID.get_secret_value(),
    aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY.get_secret_value(),
    region_name=settings.AWS_REGION,
)

ALLOWED_TYPES = {
    'image/jpeg': '.jpg',
    'image/png': '.png',
    'image/webp': '.webp',
    'application/pdf': '.pdf',
}

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB

def generate_upload_url(
    user_id: int,
    content_type: str,
    original_filename: str,
) -> dict:
    """Generate a presigned URL for direct browser-to-S3 upload."""

    if content_type not in ALLOWED_TYPES:
        raise BadRequestException(
            f"File type '{content_type}' is not allowed. "
            f"Accepted types: {', '.join(ALLOWED_TYPES.keys())}"
        )

    # Generate a safe, unique filename
    extension = ALLOWED_TYPES[content_type]
    date_prefix = datetime.utcnow().strftime('%Y/%m/%d')
    safe_name = f"{date_prefix}/{user_id}/{uuid.uuid4()}{extension}"

    # Generate presigned URL
    presigned = s3_client.generate_presigned_url(
        'put_object',
        Params={
            'Bucket': settings.S3_BUCKET_NAME,
            'Key': safe_name,
            'ContentType': content_type,
            'ContentLength': MAX_FILE_SIZE,  # Enforce max size
        },
        ExpiresIn=300,  # URL valid for 5 minutes
    )

    return {
        'upload_url': presigned,
        'file_key': safe_name,
        'expires_in': 300,
    }

Step 2: The API Endpoint

# app/uploads/router.py
from pydantic import BaseModel, Field

class UploadRequest(BaseModel):
    content_type: str = Field(..., description="MIME type of the file")
    filename: str = Field(..., min_length=1, max_length=255)

class UploadResponse(BaseModel):
    upload_url: str
    file_key: str
    expires_in: int

@router.post("/uploads/request-url", response_model=UploadResponse)
def request_upload_url(
    data: UploadRequest,
    user: User = Depends(get_current_user),
):
    result = storage_service.generate_upload_url(
        user_id=user.id,
        content_type=data.content_type,
        original_filename=data.filename,
    )
    return result

Step 3: Frontend Uploads Directly to S3

// Frontend: request URL, then upload directly to S3
async function uploadFile(file) {
    // 1. Get presigned URL from your API
    const response = await fetch('/api/uploads/request-url', {
        method: 'POST',
        headers: {
            'Authorization': `Bearer ${token}`,
            'Content-Type': 'application/json',
        },
        body: JSON.stringify({
            content_type: file.type,
            filename: file.name,
        }),
    });
    const { upload_url, file_key } = await response.json();

    // 2. Upload directly to S3 — bypasses your API completely
    await fetch(upload_url, {
        method: 'PUT',
        headers: { 'Content-Type': file.type },
        body: file,
    });

    // 3. Confirm upload to your API
    await fetch('/api/uploads/confirm', {
        method: 'POST',
        headers: {
            'Authorization': `Bearer ${token}`,
            'Content-Type': 'application/json',
        },
        body: JSON.stringify({ file_key }),
    });

    return file_key;
}

The browser sends the file directly to S3. Your API server never sees the file data — it only generates the URL and records the metadata. A 200MB upload doesn't consume any memory on your API server.

Step 4: Confirm and Record the Upload

None

After the browser uploads to S3, it notifies your API:

# app/uploads/router.py

class ConfirmUploadRequest(BaseModel):
    file_key: str

@router.post("/uploads/confirm")
def confirm_upload(
    data: ConfirmUploadRequest,
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    # Verify the file exists in S3
    try:
        head = s3_client.head_object(
            Bucket=settings.S3_BUCKET_NAME,
            Key=data.file_key,
        )
    except s3_client.exceptions.NoSuchKey:
        raise BadRequestException("File not found in storage")

    # Verify ownership (file key starts with user's ID path)
    if f"/{user.id}/" not in data.file_key:
        raise ForbiddenException("You don't own this file")

    # Record in database
    file_record = FileUpload(
        user_id=user.id,
        file_key=data.file_key,
        content_type=head['ContentType'],
        size_bytes=head['ContentLength'],
        status='confirmed',
    )
    db.add(file_record)
    db.commit()

    # Generate download URL
    download_url = generate_download_url(data.file_key)

    return {
        "file_key": data.file_key,
        "download_url": download_url,
        "size_bytes": head['ContentLength'],
    }

The confirmation step verifies the file actually exists in S3, checks ownership, and records the metadata. This two-phase approach (request URL → upload → confirm) ensures you only track files that were actually uploaded.

When You Still Need Server-Side Uploads

None

Presigned URLs are the right default, but some cases genuinely require your server to handle the file:

Files that need server-side processing before storage. Image resizing, PDF text extraction, virus scanning — anything that transforms the file before storing it.

Files from non-browser clients. Celery tasks, management commands, or internal services that generate files don't need presigned URLs. They can use boto3 directly.

Very small files where the presigned URL overhead isn't worth it. A 5KB avatar upload doesn't benefit from the presigned URL dance. The overhead of two API calls (request URL + confirm) exceeds the cost of just receiving the file directly.

For these cases, handle uploads safely:

# app/uploads/router.py
from fastapi import UploadFile, File
import hashlib

@router.post("/uploads/direct")
async def direct_upload(
    file: UploadFile = File(...),
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    # Validate content type
    if file.content_type not in ALLOWED_TYPES:
        raise BadRequestException(f"File type {file.content_type} not allowed")

    # Read in chunks — never load entire file into memory
    contents = bytearray()
    file_hash = hashlib.sha256()
    total_size = 0

    while chunk := await file.read(1024 * 64):  # 64KB chunks
        total_size += len(chunk)
        if total_size > MAX_FILE_SIZE:
            raise BadRequestException("File exceeds maximum size of 10MB")
        contents.extend(chunk)
        file_hash.update(chunk)

    # Verify content type matches actual file content
    actual_type = detect_content_type(bytes(contents[:2048]))
    if actual_type != file.content_type:
        raise BadRequestException(
            f"File content doesn't match declared type. "
            f"Expected {file.content_type}, got {actual_type}"
        )

    # Generate safe key
    extension = ALLOWED_TYPES[file.content_type]
    date_prefix = datetime.utcnow().strftime('%Y/%m/%d')
    file_key = f"{date_prefix}/{user.id}/{uuid.uuid4()}{extension}"

    # Upload to S3 from server
    s3_client.put_object(
        Bucket=settings.S3_BUCKET_NAME,
        Key=file_key,
        Body=bytes(contents),
        ContentType=file.content_type,
    )

    # Record in database
    file_record = FileUpload(
        user_id=user.id,
        file_key=file_key,
        content_type=file.content_type,
        size_bytes=total_size,
        file_hash=file_hash.hexdigest(),
        status='confirmed',
    )
    db.add(file_record)
    db.commit()

    return {"file_key": file_key, "size": total_size}

The chunked reading (while chunk := await file.read(64 * 1024)) is critical. It prevents loading a 200MB file entirely into memory. Each 64KB chunk is processed and accumulated. If the total exceeds the limit, the upload is rejected immediately without reading the rest.

The content type verification — checking the actual file bytes against the declared MIME type — prevents the .php file renamed to .jpg attack from my opening story.

Generating Secure Download URLs

None

Never serve files by making them publicly accessible. Generate time-limited download URLs:

def generate_download_url(file_key: str, expires_in: int = 3600) -> str:
    """Generate a presigned URL for downloading a file."""
    return s3_client.generate_presigned_url(
        'get_object',
        Params={
            'Bucket': settings.S3_BUCKET_NAME,
            'Key': file_key,
        },
        ExpiresIn=expires_in,
    )

@router.get("/files/{file_id}/download")
def get_download_url(
    file_id: int,
    user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    file_record = db.query(FileUpload).filter(
        FileUpload.id == file_id,
        FileUpload.user_id == user.id,  # Ownership check
    ).first()

    if not file_record:
        raise NotFoundException("File", file_id)

    url = generate_download_url(file_record.file_key)
    return {"download_url": url, "expires_in": 3600}

The download URL expires after one hour. After that, it's useless. The user must request a new URL from your API, which re-verifies their authentication and authorization. This means you can revoke access to files instantly by changing permissions in your API — even though the file itself is in S3.

Cleaning Up Orphaned Files

None

Files that were never confirmed (the user closed the browser mid-upload) accumulate in S3. Clean them up with a periodic task:

# tasks/cleanup.py
@shared_task
def cleanup_orphaned_uploads():
    """Remove S3 objects that were never confirmed in the database."""
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')

    cutoff = datetime.utcnow() - timedelta(hours=24)
    deleted_count = 0

    for page in paginator.paginate(Bucket=settings.S3_BUCKET_NAME):
        for obj in page.get('Contents', []):
            # Skip recently uploaded files
            if obj['LastModified'].replace(tzinfo=None) > cutoff:
                continue

            # Check if this file has a confirmed record in the database
            exists = db.query(FileUpload).filter(
                FileUpload.file_key == obj['Key'],
                FileUpload.status == 'confirmed',
            ).first()

            if not exists:
                s3.delete_object(
                    Bucket=settings.S3_BUCKET_NAME,
                    Key=obj['Key'],
                )
                deleted_count += 1

    logger.info(f"Cleaned up {deleted_count} orphaned files")

Run this daily via Celery Beat. The 24-hour cutoff gives users time to complete their uploads before cleanup runs.

The S3 Bucket Configuration

None

A few settings that prevent common problems:

# Set up via boto3 or Terraform/CloudFormation

# CORS — required for browser-to-S3 uploads
cors_configuration = {
    'CORSRules': [{
        'AllowedHeaders': ['*'],
        'AllowedMethods': ['PUT', 'GET'],
        'AllowedOrigins': [
            'https://yourdomain.com',
            'http://localhost:3000',  # Development
        ],
        'MaxAgeSeconds': 3600,
    }]
}

# Lifecycle rule — auto-delete unconfirmed uploads after 7 days
lifecycle_configuration = {
    'Rules': [{
        'ID': 'cleanup-incomplete-uploads',
        'Status': 'Enabled',
        'Filter': {'Prefix': ''},
        'AbortIncompleteMultipartUpload': {
            'DaysAfterInitiation': 7,
        },
    }]
}

# Block all public access — files served via presigned URLs only
public_access_block = {
    'BlockPublicAcls': True,
    'IgnorePublicAcls': True,
    'BlockPublicPolicy': True,
    'RestrictPublicBuckets': True,
}

BlockPublicAcls: True is the most important setting. It ensures no file in your bucket is ever publicly accessible. Every download must go through a presigned URL generated by your API, which means every download is authenticated and authorized.

Bottom Line

File uploads are not a feature. They're a distributed systems problem disguised as a form field. The moment you accept file data through your API server, you inherit memory management, disk management, security validation, and scalability challenges that have nothing to do with your application logic.

The presigned URL pattern solves this by removing your API from the file transfer path entirely. The browser uploads directly to S3. Your API only handles metadata — generating URLs, recording file records, enforcing permissions. This is the pattern used by every large-scale file upload system — Slack, Dropbox, GitHub — because it scales independently of your API server.

Those three production failures from my opening story — memory exhaustion, disk full, malicious file upload — are all prevented by this architecture. Your API never holds file data in memory. Files live in S3, which has virtually unlimited storage. Content type validation happens before the upload URL is generated. Three problems, one architecture, zero file data touching your server.

How do you handle file uploads in your projects? I'm curious about the split — are you still using server-side uploads, or have you moved to presigned URLs? And for those handling really large files (100MB+), do you use multipart uploads? Share your setup in the comments.

A special thanks to Level Up Coding for giving writers and engineers a space to share practical, real-world lessons like this. I'm grateful for the opportunity to publish this piece with the publication and contribute to a community that cares about better engineering.

None