Handle FastAPI file uploads with size limits and safe storage

FastAPI's UploadFile gives an endpoint a convenient interface to multipart uploads. A production upload path still needs limits on request size, validation of the actual content, server-generated storage names, and a private destination. These controls protect different stages of the request; a size check inside the endpoint cannot protect resources already consumed while parsing the upload.

This guide starts with one file, adds bounded copying and cleanup, then handles a small batch. The examples illustrate local storage. Authentication, quotas, scanning, and download authorization remain application responsibilities.

Start with a small-file endpoint

Install FastAPI and the multipart parser:

pip install "fastapi[standard]" python-multipart

from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload(file: UploadFile):
    content = await file.read()
    return {"filename": file.filename, "size": len(content)}

UploadFile uses a spooled temporary file: smaller content can remain in memory, with larger content moved to disk. Calling await file.read() without a size still creates an in-memory copy of the entire file. Use that form only for uploads whose size is bounded before they reach the handler. See FastAPI's file-upload guide.

Multipart parsing generally happens before FastAPI calls the route. Copying the temporary file in chunks reduces the handler's additional memory use, but it does not turn this endpoint into a direct stream from the client or replace an ingress request-body limit.

Validate content and write to a private directory

A file's extension and client-supplied Content-Type are hints, not proof of its contents. The following replacement application uses python-magic to screen the file signature, allows three MIME types, and caps each saved file at 10 MiB.

Install the Python wrapper and the operating system's libmagic library. The python-magic installation guide lists platform-specific requirements; installing the Python package alone may be insufficient.

pip install python-magic

Replace the first application with this one; do not register two handlers for the same POST /upload route:

from pathlib import Path
from uuid import uuid4

import magic
from fastapi import FastAPI, HTTPException, UploadFile

app = FastAPI()
MIME_SUFFIX = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "application/pdf": ".pdf",
}
MAX_SIZE = 10 * 1024 * 1024
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)

def save_one(file: UploadFile):
    path = None
    try:
        head = file.file.read(2048)
        mime = magic.from_buffer(head, mime=True)
        if mime not in MIME_SUFFIX:
            raise HTTPException(415, "Unsupported file type")
        file.file.seek(0)
        safe_name = f"{uuid4().hex}{MIME_SUFFIX[mime]}"
        total = 0
        candidate = UPLOAD_DIR / safe_name
        # Exclusive creation avoids replacing an existing file.
        with candidate.open("xb") as output:
            path = candidate
            while chunk := file.file.read(1024 * 1024):
                total += len(chunk)
                if total > MAX_SIZE:
                    raise HTTPException(413, "File exceeds the size limit")
                output.write(chunk)
        return {"saved": safe_name, "size": total}
    except BaseException:
        # Remove only a file successfully created by this operation.
        if path is not None:
            path.unlink(missing_ok=True)
        raise
    finally:
        file.file.close()

@app.post("/upload")
def upload_secure(file: UploadFile):
    return save_one(file)

A synchronous def route runs in FastAPI's thread pool, so the synchronous signature check and file writes do not run directly on the event loop. This still consumes worker, CPU, and storage capacity; limit concurrent uploads as well as individual file sizes.

The code deliberately never joins file.filename onto a storage path. A name such as ../../etc/passwd becomes display metadata only, while a UUID determines the actual destination. Escape the original name if displaying it in HTML, and do not use it as an authorization identifier.

mode=0o700 applies when the directory is created on a platform supporting those permissions. It does not repair the permissions of an existing directory. Provision and inspect the storage location separately, keep it outside the public web root, and do not grant execute permissions to uploaded files.

Signature detection is a screening step. A file that resembles a PNG or PDF can still contain dangerous or malformed content. Validate it with the appropriate parser under resource limits, apply scanning or content reconstruction where required, and keep it unavailable until processing succeeds. The OWASP file-upload guidance explains why several controls are necessary.

Limit batches and clean up partial results

Add this endpoint to the replacement application:

@app.post("/upload/batch")
def upload_batch(files: list[UploadFile]):
    if not 1 <= len(files) <= 5:
        for file in files:
            file.file.close()
        raise HTTPException(400, "Upload between one and five files")
    results = []
    try:
        for file in files:
            results.append(save_one(file))
        return {"saved": results}
    except BaseException:
        for result in results:
            (UPLOAD_DIR / result["saved"]).unlink(missing_ok=True)
        raise
    finally:
        for file in files:
            file.file.close()

If a later file fails validation, the handler removes earlier files saved by that batch. The file that failed is cleaned up by save_one. This is best-effort local cleanup, not a transaction that survives process termination, disk errors, or a host failure. Maintain a reconciliation task for orphaned uploads.

Five files at 10 MiB each permit up to 50 MiB of file content. The HTTP body also contains multipart boundaries and headers. Set a total request-body limit at the gateway that accounts for that overhead, and configure parser limits appropriate to your framework version. The route's file-count check happens after parsing; it is not a parser resource limit.

Separate upload progress from processing progress

A browser upload-progress event reports bytes sent. It does not tell the user that the server has validated, scanned, transformed, or durably stored the content. For large files, expose separate states such as uploading, processing, accepted, and rejected.

Move expensive scanning, transcoding, and thumbnail generation into an appropriate worker system. Persist the file and its job record before acknowledging acceptance. Do not pass the request's temporary UploadFile object to work that may run after the request has closed it. If the work must survive restarts, use durable storage and a persistent queue.

Images may need dimension limits and decode/re-encode processing; decompression can consume much more memory than the uploaded byte count. Documents need their own scanning and download policies. Private object storage still requires authorization, expiration, lifecycle rules, and safe download headers. It does not make an unsafe file harmless.

Add per-user quotas, rate limits, concurrency limits, and disk monitoring so many individually valid uploads cannot exhaust the service.

Test rejection and cleanup paths

Use FastAPI's TestClient with a multipart file:

import io
from fastapi.testclient import TestClient

client = TestClient(app)
# png_bytes must contain an actual PNG fixture, not just a .png filename.
response = client.post(
    "/upload",
    files={"file": ("sample.png", io.BytesIO(png_bytes), "image/png")},
)
assert response.status_code == 200

Test a valid file, an unsupported format, an oversized file, a traversal-style filename, a batch where a later file fails, and a simulated storage failure. At the deployment boundary, also test an interrupted request and the gateway's total-body limit. Check that rejected requests leave no newly created application files and that accepted files stay in the intended directory.

Record the result, file identifier, byte count, and elapsed time. Avoid logging file contents, credentials, or signed download URLs. If Guance already collects the service's logs, use the DataKit file collector to inspect rejection and storage-failure events. Verify a known failure before using those fields in an alert.

Common questions

Do I still need a limit such as NGINX's client_max_body_size?

Yes. The gateway can reject an oversized request before application parsing consumes resources. The handler enforces the application's per-file rule. Configure and test both layers.

Who cleans up temporary files?

FastAPI manages request-upload resources, and the examples close each file explicitly after use. Files copied into your own directory or object storage have a separate lifecycle. Remove failed or abandoned uploads and define retention for accepted files.

Does chunked copying make arbitrarily large uploads safe?

No. It bounds one copy buffer, but multipart parsing, temporary disk usage, worker availability, timeouts, and total storage remain constraints. For very large objects, evaluate a dedicated upload path with explicit limits and authorized object-storage access.

Related reading

• Tortoise ORM vs. SQLAlchemy (Chinese)
• Managing Python packages, projects, and interpreters with uv (Chinese)