File Uploads and Streaming Responses in FastAPI

File Uploads in FastAPI

FastAPI handles file uploads through Python’s UploadFile class, which wraps the uploaded file with an async interface. Unlike raw form data, UploadFile uses a spooled temporary file — small files stay in memory, large files spill to disk automatically.

Basic File Upload

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

app = FastAPI()

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

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    file_path = UPLOAD_DIR / file.filename
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file_path.stat().st_size
    }

File(...) marks the parameter as required. The uploaded file is available through file.file (a file-like object), file.filename, and file.content_type.

Uploading Multiple Files

from typing import List

@app.post("/upload-multiple")
async def upload_multiple(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        file_path = UPLOAD_DIR / file.filename
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        results.append({
            "filename": file.filename,
            "size": file_path.stat().st_size
        })
    return {"uploaded": len(results), "files": results}

File Validation

Always validate uploads. Never trust client-provided filenames or content types.

from fastapi import HTTPException
import uuid

ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp", "application/pdf"}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

def sanitize_filename(filename: str) -> str:
    """Generate a safe filename while preserving the extension."""
    ext = Path(filename).suffix.lower()
    return f"{uuid.uuid4().hex}{ext}"

async def validate_file(file: UploadFile, allowed_types: set, max_size: int):
    """Validate file type and size."""
    if file.content_type not in allowed_types:
        raise HTTPException(
            status_code=400,
            detail=f"File type {file.content_type} not allowed. "
                   f"Allowed: {', '.join(allowed_types)}"
        )

    # Read file to check size
    contents = await file.read()
    if len(contents) > max_size:
        raise HTTPException(
            status_code=400,
            detail=f"File too large. Maximum size: {max_size // (1024*1024)} MB"
        )

    # Reset file position for later reading
    await file.seek(0)
    return contents

@app.post("/upload-validated")
async def upload_validated(file: UploadFile = File(...)):
    contents = await validate_file(file, ALLOWED_TYPES, MAX_FILE_SIZE)

    safe_name = sanitize_filename(file.filename)
    file_path = UPLOAD_DIR / safe_name

    with open(file_path, "wb") as f:
        f.write(contents)

    return {
        "original_name": file.filename,
        "saved_as": safe_name,
        "size": len(contents),
        "content_type": file.content_type
    }

Key points: never use the original filename directly on disk (use UUIDs), always validate the content type, and check the file size before processing.

Chunked Upload for Large Files

For very large files, avoid loading the entire file into memory. Read and write in chunks.

@app.post("/upload-large")
async def upload_large_file(file: UploadFile = File(...)):
    safe_name = sanitize_filename(file.filename)
    file_path = UPLOAD_DIR / safe_name

    total_size = 0
    chunk_size = 1024 * 1024  # 1 MB chunks

    with open(file_path, "wb") as f:
        while True:
            chunk = await file.read(chunk_size)
            if not chunk:
                break
            total_size += len(chunk)
            if total_size > MAX_FILE_SIZE:
                # Clean up partial file
                f.close()
                file_path.unlink(missing_ok=True)
                raise HTTPException(400, "File exceeds maximum size")
            f.write(chunk)

    return {
        "saved_as": safe_name,
        "size_bytes": total_size
    }

This approach uses constant memory regardless of file size. The size check happens during upload, so you stop early rather than wasting bandwidth.

File Upload with Additional Form Data

Combine file uploads with regular form fields.

from fastapi import Form

@app.post("/documents")
async def upload_document(
    file: UploadFile = File(...),
    title: str = Form(...),
    description: str = Form(""),
    category: str = Form("general")
):
    safe_name = sanitize_filename(file.filename)
    file_path = UPLOAD_DIR / safe_name

    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {
        "title": title,
        "description": description,
        "category": category,
        "file": {
            "saved_as": safe_name,
            "original_name": file.filename,
            "content_type": file.content_type
        }
    }

When mixing files and form data, use Form() for regular fields instead of Body(). The request must be sent as multipart/form-data.

Streaming Responses

Streaming responses send data in chunks rather than building the entire response in memory. This is essential for large files, real-time data, and generated content.

Streaming a File Download

from fastapi.responses import StreamingResponse, FileResponse
from pathlib import Path

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = UPLOAD_DIR / filename
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")

    # For known files, FileResponse is simpler and handles caching
    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream"
    )

FileResponse handles Content-Length, Content-Disposition, and supports range requests for resumable downloads.

Streaming Generated Data

For data that is generated on the fly — CSV exports, log streams, or chunked API responses — use StreamingResponse with a generator.

import csv
import io
from datetime import datetime

async def generate_csv_report(report_id: int):
    """Generate CSV data row by row."""
    # Header row
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["id", "timestamp", "value", "status"])
    yield output.getvalue()
    output.seek(0)
    output.truncate(0)

    # Data rows - fetched in batches
    offset = 0
    batch_size = 1000
    while True:
        rows = await fetch_report_rows(report_id, offset, batch_size)
        if not rows:
            break
        for row in rows:
            writer.writerow([row.id, row.timestamp, row.value, row.status])
        yield output.getvalue()
        output.seek(0)
        output.truncate(0)
        offset += batch_size

@app.get("/reports/{report_id}/csv")
async def download_report(report_id: int):
    return StreamingResponse(
        generate_csv_report(report_id),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=report_{report_id}.csv"
        }
    )

The generator yields chunks of data. FastAPI sends each chunk to the client as it is produced, so memory usage stays constant even for million-row exports.

Server-Sent Events (SSE)

Stream real-time updates to the browser using SSE. Unlike WebSockets, SSE is unidirectional (server to client) and works over standard HTTP.

import asyncio
import json

async def event_stream(channel: str):
    """Generate SSE events."""
    while True:
        # Check for new events (from a queue, database, etc.)
        event = await get_next_event(channel)
        if event:
            data = json.dumps(event)
            yield f"event: {event['type']}\ndata: {data}\n\n"
        else:
            # Send a keep-alive comment every 15 seconds
            yield ": keepalive\n\n"
            await asyncio.sleep(15)

@app.get("/events/{channel}")
async def stream_events(channel: str):
    return StreamingResponse(
        event_stream(channel),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )

Streaming a Large JSON Response

For endpoints that return large JSON arrays, stream the response to avoid building the entire list in memory.

import json

async def stream_items():
    yield "["
    first = True
    offset = 0
    batch_size = 100

    while True:
        items = await fetch_items_batch(offset, batch_size)
        if not items:
            break
        for item in items:
            if not first:
                yield ","
            yield json.dumps(item.dict())
            first = False
        offset += batch_size

    yield "]"

@app.get("/items/export")
async def export_all_items():
    return StreamingResponse(
        stream_items(),
        media_type="application/json"
    )

Uploading to Cloud Storage

In production, you typically upload to S3, GCS, or Azure Blob rather than local disk.

import boto3
from botocore.exceptions import ClientError

s3_client = boto3.client("s3")
BUCKET_NAME = "my-app-uploads"

@app.post("/upload-s3")
async def upload_to_s3(file: UploadFile = File(...)):
    safe_name = sanitize_filename(file.filename)
    key = f"uploads/{safe_name}"

    try:
        s3_client.upload_fileobj(
            file.file,
            BUCKET_NAME,
            key,
            ExtraArgs={
                "ContentType": file.content_type,
                "ServerSideEncryption": "AES256"
            }
        )
    except ClientError as e:
        raise HTTPException(500, f"Upload failed: {str(e)}")

    url = f"https://{BUCKET_NAME}.s3.amazonaws.com/{key}"
    return {"url": url, "key": key}

Key Takeaways

FastAPI’s UploadFile handles file uploads efficiently with spooled temporary files. Always validate file types and sizes before processing, use sanitized filenames, and read large files in chunks to control memory usage. For responses, use FileResponse for static files and StreamingResponse with generators for dynamic or large content. In production, stream uploads directly to cloud storage rather than saving to local disk.