Skip to main content
友田 陽大
Python backend
Python
FastAPI
ファイルアップロード
ストリーミング
セキュリティ

FastAPI File Uploads, Forms, and Streaming Production Guide: Handling UploadFile / Form / StreamingResponse Safely and Idempotently Without Exhausting Memory

A guide to handling FastAPI file uploads (UploadFile/File), forms (Form), and streaming (StreamingResponse/FileResponse) at production quality. Explained with real code: the design judgment of going straight to object storage with a presigned URL, avoiding memory exhaustion with chunk reads, and size limits, MIME/magic-byte validation, path-traversal defense, malware scanning, and content-hash idempotency.

Published
Reading time
22 min read
Author
友田 陽大
Share

"I want to add file upload to FastAPI" — the requirement is one line. But file reception is a domain where one line's mistake directly becomes the server's memory exhaustion, disk destruction, or arbitrary-file write. Loading a 5GiB video whole into memory with await file.read() and the process dies; using the uploaded filename as a path as-is and ../../etc/passwd gets written; trusting the Content-Type and saving an executable as an image — all of these, because "it works with a small file," go unnoticed in production until a user throws a big one.

This article is a guide for implementing production-quality file upload, form processing, and streaming responses in FastAPI. While following the official FastAPI tutorial's UploadFile / Form / StreamingResponse faithfully to the latest official spec, I step into the domain the official docs (being teaching material) don't touch — how to receive large files without exhausting memory, enforcing size limits, MIME/magic-byte validation, path-traversal defense, malware scanning, idempotency, and the design judgment of "should you receive it via the API at all." As a subject, I interweave the judgments from an in-house AI platform I built for a major domestic broadcaster (a monorepo bundling multiple FastAPI services; I designed a zero-trust entrance that malware-scans up to GiB-class video/image material brought in from outside, before it reaches the platform, and sorts it into clean/quarantine).

The rule of this article: the APIs / recommended libraries are based on the FastAPI official documentation (as of June 2026). Receiving files/forms requires python-multipart (officially stated). Specs get revised, so always confirm the latest behavior officially before shipping to production. Secrets (bucket name, signing key, connection info) presuppose environment variables (hardcoding strictly forbidden). Files arriving from outside are all untrusted — this is the rule running through this article.


0. First, the Judgment: Receive via the API, or Send Straight with a Presigned URL

Before entering the implementation, there's the first branch you should decide in production design. "Does that file really need to be received through the FastAPI process?" Skip this and receive everything via the API, and the API's memory, bandwidth, CPU, and cost get eaten on every upload, becoming a shackle on scale.

There are 2 options.

  • (A) Receive via the API: client → FastAPI → storage. FastAPI relays the byte sequence. Easy to validate / transform on the spot.
  • (B) Go straight with a presigned URL: FastAPI only issues "a presigned URL for the upload destination." The client PUTs directly to object storage (S3 / GCS). The byte sequence doesn't pass through the API.
Viewpoint(A) Receive via the API(B) Go straight with a presigned URL
Suited caseSmall files (~a few MB), want to validate/transform on the spotLarge (tens of MB to GiB), image/video/backup
The API's memory/bandwidthAll bytes pass through the API (a load source)Don't pass through the API (the process stays light)
ScaleThe API's throughput is the capStorage scales (the API is irrelevant)
CostThe API's execution time / transfer piles upThe API only issues URLs = cheap
Validation timingOn the spot at receptionAfter upload completion, event-driven (after the fact)
Implementation simplicitySimple (KISS)Needs the wiring of signing, CORS, completion notification

The criterion: if it could exceed a few MB, first consider (B) presigned-URL direct. "Just receive everything via the API" works while it's small, but the moment a user throws a big one, the process drops with memory exhaustion. This article centers on how to do (A) correctly and safely, while also showing (B)'s design (Chapter 7) concretely. Many production systems' realistic solution is a hybrid of "small files via A, large via B."

In the broadcaster platform, I sent the upload of up to GiB-class material straight to storage with presigned-URL chunk parallelism (up to 8 parallel), not letting the API process relay the byte sequence. This is the core of the entrance design that "doesn't exhaust memory even at GiB-class."


1. UploadFile and bytes (File): Which to Receive With

Once you decide to receive via the API (option A), the first judgment is receive with UploadFile or with bytes (File). To say the conclusion first, production is, as a rule, UploadFile. The reason is clear.

First, line up both's minimal code, per the official docs. Reception requires python-multipart (because an uploaded file is sent as "form data").

pip install python-multipart
from typing import Annotated
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

# (1) bytes で受ける:ファイルの中身が"丸ごとメモリ"に載る
@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}

# (2) UploadFile で受ける:スプールされたファイルオブジェクトを受け取る
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

1.1 The Advantages of UploadFile (the 4 Points the Official Docs List)

The official docs clearly enumerate why UploadFile is superior to bytes.

  • You don't have to write File() as a default value (it holds with just the type annotation).
  • It uses a "spooled" filekeeps it in memory up to a certain size, and evacuates to disk above that threshold.
  • This lets you handle images, videos, and large binaries without eating up memory.
  • You can get metadata from the uploaded file.
  • It has a file-like async interface.

In contrast, bytes (File) reads the file's contents whole into memory. len(file) is immediately gettable and handy, but at large sizes it instantly exhausts memory. So bytes should be limited to "files of a few KB to at most a few hundred KB, which you're sure are small."

The point: bytes's handiness is the handiness of "only when you can guarantee it's small." The upload size is decided by the client, and things exceeding your assumption will surely come. When in doubt, UploadFile — the spool structurally prevents memory exhaustion.

1.2 UploadFile's Metadata and async Methods

UploadFile has the following metadata.

  • filename: the upload-source file name (e.g. myimage.jpg). A value you must not trust (sanitize in Chapter 4).
  • content_type: the MIME type (e.g. image/jpeg). This too is self-declared and not over-trusted (validate in Chapter 5).
  • file: a SpooledTemporaryFile (a file-like object).

And the operations are async methods. Don't forget await.

contents = await file.read()      # size バイト読む(引数なしで全部)
await file.write(data)            # data(str または bytes)を書く
await file.seek(offset)           # offset バイト位置へ移動
await file.close()                # 閉じる

Don't call await file.read() unconditionally: the argument-less await file.read() expands all bytes into memory. This is the same memory consumption as receiving with bytes — the act of discarding UploadFile's spool benefit yourself. If you anticipate large sizes, chunk-read with read(size) (Chapter 3). It's not "safe because I use UploadFile" but "safe because I read in chunks."


2. Single File and Multiple Files

2.1 Optional Upload

If you don't make the file required, allow None per the official docs.

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No upload file sent"}
    return {"filename": file.filename}

2.2 Multiple Files Are list[UploadFile]

To receive multiple files, just make the type list[UploadFile].

from typing import Annotated
from fastapi import File, UploadFile

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}

# 説明(description)などのメタデータを足すなら File() を併用する
@app.post("/uploadfiles-described/")
async def create_upload_files_described(
    files: Annotated[
        list[UploadFile], File(description="Multiple files as UploadFile")
    ],
):
    return {"filenames": [file.filename for file in files]}

Even with multiple files, think of the limit as "total": even upholding just "the size limit of 1 file," sending 10,000 small files at once eats up both memory and disk. In production, enforce both "the limit per file" and "the limit on the total byte count / file count per request" (Chapter 3).


3. Large Files Safely: Chunk Read, Size Limit, Backpressure

Here is this chapter's heart. What divides "works" from "doesn't fall over even at large sizes" is almost this one pointstream-process the byte sequence in chunks (small pieces) rather than in bulk.

3.1 NG: Load Everything into Memory

# ❌ アンチパターン:5GiB の動画でも丸ごとメモリに載る → OOM でプロセス死
@app.post("/upload-bad/")
async def upload_bad(file: UploadFile):
    contents = await file.read()        # 全バイトをメモリ展開(危険)
    with open(f"/data/{file.filename}", "wb") as f:   # ファイル名もそのまま(危険・第4章)
        f.write(contents)
    return {"size": len(contents)}

This is doubly dangerous. (1) Load all bytes into memory. (2) Use the user-specified filename as the write path as-is (path traversal).

3.2 OK: Read in Chunks, Enforce the Limit with a Running Byte Count

from fastapi import FastAPI, UploadFile, HTTPException, status

app = FastAPI()

CHUNK_SIZE = 1024 * 1024          # 1MiB ずつ読む(メモリ常駐は1チャンク分だけ)
MAX_FILE_SIZE = 200 * 1024 * 1024  # 例:1ファイル200MiB上限

async def save_within_limit(file: UploadFile, dest_path: str) -> int:
    """UploadFile をチャンクで読みつつ、上限超過なら即座に中断する。"""
    total = 0
    with open(dest_path, "wb") as out:
        # read(size) は「最大 size バイト」を返す。空(b"")になったら終端。
        while chunk := await file.read(CHUNK_SIZE):
            total += len(chunk)
            # ⚠️ 逐次カウントで上限を強制。全部受けてから測るのでは手遅れ。
            if total > MAX_FILE_SIZE:
                out.close()
                # 途中まで書いた分は破棄する(中途半端なファイルを残さない)
                raise HTTPException(
                    status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                    detail="ファイルサイズが上限を超えています",
                )
            out.write(chunk)
    return total

There are 3 points.

  • Only 1 chunk (here 1MiB) is ever in memory at a time. Even if the file is 5GiB, memory consumption is constant. This is the basic form of backpressure (process and discard only what you read).
  • Enforce the size limit with a "running byte count." Receiving everything then measuring with len() means by that point you've already eaten the memory/disk. Interrupt the moment it exceeds, while reading is the correct way.
  • When the limit is exceeded, return 413 Request Entity Too Large.

3.3 Don't Over-Trust Content-Length

You tend to think "look at the Content-Length header and reject if too large," but Content-Length is self-declared. An attacker can declare a small value and send a large amount, or not attach the header (chunked transfer).

# Content-Length は「早期拒否のヒント」としては有用だが、唯一の防御にはしない
from fastapi import Request

async def reject_obviously_too_large(request: Request, max_bytes: int) -> None:
    declared = request.headers.get("content-length")
    if declared is not None and int(declared) > max_bytes:
        raise HTTPException(status_code=413, detail="宣言サイズが上限超過")
    # ↑ あくまで"早めに弾く"だけ。実バイト数は 3.2 の逐次カウントで必ず再検証する。

Defend in layers: early rejection by Content-Length is merely an optimization to "cut a wasteful transfer early." The final truth is "the bytes actually read." Further, also cap the body size at the upstream reverse proxy (Nginx's client_max_body_size), load balancer, or WAF, blocking a huge body before it reaches the API — that's production defense in depth. For the design of decoupling heavy post-processing (transform, analysis) from the request, read the production-operations guide alongside.


4. Filename Sanitization: Preventing Path Traversal

UploadFile.filename is an untrusted string coming from outside. Use it as a save path as-is, and with path traversal like ../../etc/cron.d/evil, a file gets written to an unintended place.

The safest is to not use the client-specified filename in the save path at all. Generate a unique name on the server, and (if needed) hold the original filename only as metadata in the DB.

import uuid
from pathlib import Path

UPLOAD_DIR = Path("/data/uploads").resolve()    # 保存先の基準ディレクトリ(絶対パス)

def safe_destination(original_filename: str | None) -> Path:
    # 拡張子だけは(検証済みのものを)引き継いでよいが、名前本体はサーバーが採番する
    suffix = Path(original_filename or "").suffix.lower()
    allowed_suffix = suffix if suffix in {".png", ".jpg", ".jpeg", ".pdf", ".mp4"} else ""
    # ユーザー入力に依存しない一意名。これでパストラバーサルは原理的に起きない。
    dest = UPLOAD_DIR / f"{uuid.uuid4().hex}{allowed_suffix}"

    # 二重の防御:解決後パスが基準ディレクトリ配下にあることを必ず確認する
    resolved = dest.resolve()
    if not resolved.is_relative_to(UPLOAD_DIR):     # Python 3.9+
        raise HTTPException(status_code=400, detail="不正な保存先です")
    return resolved

"Don't use it" over "sanitize and use": there's also "sanitizing" by removing / and .. from filename, but it tends to be broken by encoding loopholes (%2e%2e, NUL bytes, Unicode normalization). Using a server-numbered unique name and keeping the original name as display metadata is the most robust and simplest (KISS). In the broadcaster platform too, I stored material in Cloud Storage while building path-traversal defense into the entrance, designing so an external-origin name isn't directly connected to the save path.


5. Validate the Contents: Extension, MIME, Magic Bytes, Malware Scanning

The filename and Content-Type are self-declared. ".jpg so it's an image" / "Content-Type: image/png so it's safe" don't hold. An attacker can make an executable claim to be photo.jpg and declare image/png. Validate by the substance (the leading bytes of the contents = the magic bytes).

5.1 Three-Stage Validation: Extension → Content-Type → Magic Bytes

# 各形式の「マジックバイト(ファイルシグネチャ)」。中身の先頭で本物か判定する。
MAGIC_SIGNATURES: dict[str, tuple[bytes, ...]] = {
    "image/png":  (b"\x89PNG\r\n\x1a\n",),
    "image/jpeg": (b"\xff\xd8\xff",),
    "application/pdf": (b"%PDF-",),
}
ALLOWED_TYPES = set(MAGIC_SIGNATURES.keys())

async def validate_content_type(file: UploadFile) -> str:
    # (1) 申告された Content-Type が許可リストにあるか(必要条件であって十分条件ではない)
    declared = (file.content_type or "").lower()
    if declared not in ALLOWED_TYPES:
        raise HTTPException(status_code=415, detail="許可されていない種類です")

    # (2) 先頭バイトを覗いて、申告と"実体"が一致するか検証する
    header = await file.read(8)        # 先頭8バイトだけ読む
    await file.seek(0)                 # ⚠️ 必ず巻き戻す(このあと本保存で先頭から読むため)
    if not any(header.startswith(sig) for sig in MAGIC_SIGNATURES[declared]):
        # 申告は image/png なのに中身が PNG でない → 偽装
        raise HTTPException(status_code=415, detail="ファイルの実体が種類と一致しません")
    return declared

What works here is await file.seek(0). After reading the leading bytes for validation, always rewind before the real save — otherwise the saved file is missing its leading 8 bytes.

Magic bytes are "the minimum," not all-powerful: a leading-signature match repels most of the spoofing, but it passes a polyglot (a file valid as multiple formats) and a file that's the correct format but malicious (a crafted PDF, an embedded script inside an image). In production, layer magic-byte validation + malware scanning. There's also the hand of more robust MIME judgment with python-magic (libmagic bindings), but whether to add an external library is requirement-dependent (YAGNI).

5.2 Malware Scanning and Clean/Quarantine Sorting

For files brought in from outside, the production standard is to malware-scan before saving/publishing, and sort into clean/quarantine. Because the scanning engine (ClamAV, etc.) is heavy and slow, running it synchronously inside the API handler is forbidden — once place the upload in an 'unscanned' zone, do the scanning with an async job / event-driven, and move it to clean/quarantine by the result.

# 概念図:アップロード直後は"検査前ゾーン"。検査はリクエストから切り離す。
# 1. /upload → 検査前バケット(または隔離ステージ)に保存し、即 202 を返す
# 2. ストレージのイベント(オブジェクト作成) → 検査ワーカー(ClamAV等)を起動
# 3. クリーン → cleanバケットへ移動 / 検出 → quarantineバケットへ隔離
# 4. 結果をDBに記録し、UIへ通知(SSE/WebSocket/ポーリング)

In the broadcaster platform, I stream-scanned video/image material brought in from outside with ClamAV (Cloud Run) before it reached the platform, sorting it into clean/quarantine buckets while avoiding memory exhaustion without buffering. The scan/move was designed to complete even for up to GiB-class material, and to be idempotent against retries by the atomicity of the move. "Never let unscanned material reach the platform's interior" — this zero-trust entrance was the premise of an AI foundation handling external material.


6. Mixing Form Data (Form) and File

Often you want to send metadata (title, category, token) along with the file. What you use then is Form.

6.1 Form Alone

from typing import Annotated
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}

Without explicitly stating Form, that argument is interpreted as a query parameter or a JSON body. To receive it as a form value, Form() is mandatory (Form is a class directly inheriting Body). Form reception also needs python-multipart.

6.2 Receive Form and File in the Same Request

Per the official docs, File and Form can mix in the same path operation.

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

app = FastAPI()

@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()],
    fileb: Annotated[UploadFile, File()],
    token: Annotated[str, Form()],
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }

6.3 [Important] Form/File and a JSON Body Can't Be Received at the Same Time

Here is a pitfall you'll surely hit in production that the official docs clearly warn about.

The official warning: you can declare multiple Form / File in one path operation, but you can't simultaneously declare a Body field received as JSON. Because the request body is encoded not as application/json but as multipart/form-data (or application/x-www-form-urlencoded). This is not a FastAPI constraint but an HTTP-protocol spec.

That is, if you want to send "a file + structured JSON metadata" in one request, you can't receive a Pydantic model as a Body. The handling in practice is one of:

  • Send the metadata too as flat Form fields (title: Annotated[str, Form()], etc.). This suffices if the nesting is shallow.
  • Receive structured data as a JSON string via Form, and flow it into Pydantic inside the handler (validate at the boundary).
import json
from pydantic import BaseModel, ValidationError

class AssetMeta(BaseModel):
    title: str
    tags: list[str] = []

@app.post("/assets/")
async def create_asset(
    file: UploadFile,
    meta_json: Annotated[str, Form()],     # 構造化メタデータは JSON 文字列としてフォームで受ける
):
    try:
        # 文字列を境界で必ず検証してから内部の型へ。生の dict を信用しない。
        meta = AssetMeta.model_validate_json(meta_json)
    except ValidationError as exc:
        raise HTTPException(status_code=422, detail=exc.errors())
    return {"filename": file.filename, "title": meta.title}

The design implication: this HTTP constraint is also the reason Chapter 0 recommends "large files via presigned-URL direct." With (B), you can separate the "metadata-creation API (JSON)" and the "file body (straight to storage)" into different requests, so unbound by the multipart constraint, the API can handle clean JSON only.


7. Streaming Responses: Return Large Results and Downloads Sequentially

So far has been the "receive" side. On the "return" side too, building a large data whole in memory then returning it exhausts memory. FastAPI provides tools to return sequentially.

7.1 StreamingResponse: Send Sequentially with a Generator

StreamingResponse sends the byte sequence a (async) generator yields, while generating it. Even for a huge response, only 1 chunk is ever in memory.

import anyio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"
        await anyio.sleep(0)        # 協調的に他タスクへ譲る(キャンセルを正しく扱うため)

@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())

As a practical example, here's a form that chunk-reads a large file on storage and streams it as-is (a proxy-like download).

from fastapi.responses import StreamingResponse

def file_chunk_iterator(path: str, chunk_size: int = 1024 * 1024):
    # ファイルを1MiBずつ読んで yield。全体をメモリに載せない。
    with open(path, "rb") as f:
        while chunk := f.read(chunk_size):
            yield chunk

@app.get("/download/{asset_id}")
async def download(asset_id: str):
    path = resolve_asset_path(asset_id)         # 内部で第4章のパス検証を通す
    return StreamingResponse(
        file_chunk_iterator(path),
        media_type="application/octet-stream",
        headers={"Content-Disposition": 'attachment; filename="asset.bin"'},
    )

Backpressure and cancellation: StreamingResponse's true value is that generation proceeds at the speed the client can receive (backpressure). So you can stop generation when the client disconnects, put an appropriate await (anyio.sleep(0), etc.) inside the async generator to handle cancellation cooperatively. It's also effective in scenes where "return after it's all done" is too slow, like an LLM's sequential generation or log tailing.

7.2 FileResponse: The Standard for Returning an On-Disk File

If you just "return an existing file on the server," FileResponse is more concise and safe than writing a generator yourself. It reads the file asynchronously and auto-attaches Content-Length / Last-Modified / ETag.

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

@app.get("/report")
async def get_report():
    # path・filename・media_type を渡せる。filename は Content-Disposition に載る。
    return FileResponse(
        path="/data/reports/2026-06.pdf",
        media_type="application/pdf",
        filename="monthly-report.pdf",
    )

Use response_class=FileResponse and you can directly return a file path from a path operation too.

@app.get("/video", response_class=FileResponse)
async def get_video() -> str:
    return "/data/videos/intro.mp4"     # パスを返すだけで FileResponse が処理する

When to use StreamingResponse vs FileResponse: to return a file already on disk, FileResponse (it even handles ETagLast-Modified). For a byte sequence you generate dynamically / fetch sequentially from outside, StreamingResponse. To just return small raw data of an arbitrary media_type, Response(content=..., media_type=...) suffices (don't over-engineer).


8. Idempotency: Eliminate Duplicate Uploads with a content-hash

Upload is an operation over the network, and retries are inevitable. The client times out and resends, and the same file is saved twice — this is an accident. In production, make it idempotent.

The standard is to make the file's content-hash a unique key. Treat the same contents as the same hash = the same object, and from the 2nd time on skip the save and return the existing reference.

import hashlib
from fastapi import UploadFile, HTTPException

async def save_idempotent(file: UploadFile) -> dict:
    hasher = hashlib.sha256()
    tmp_path = make_temp_path()
    total = 0
    # チャンク読みしながら、保存とハッシュ計算を同時に進める(二度読みしない)
    with open(tmp_path, "wb") as out:
        while chunk := await file.read(1024 * 1024):
            total += len(chunk)
            if total > MAX_FILE_SIZE:
                raise HTTPException(status_code=413, detail="サイズ超過")
            hasher.update(chunk)        # ハッシュも逐次更新
            out.write(chunk)
    digest = hasher.hexdigest()

    existing = lookup_asset_by_hash(digest)     # 同じ中身が既にあるか
    if existing is not None:
        discard_temp(tmp_path)                  # 重複なので一時ファイルは捨てる
        return {"asset_id": existing.id, "deduplicated": True}

    # 最終配置は"原子的な move"で行う(書きかけが見えない・再試行に冪等)
    asset = promote_temp_to_permanent(tmp_path, key=digest)
    return {"asset_id": asset.id, "deduplicated": False}

In the broadcaster platform too, I designed long-running jobs and material moves to be idempotent against retries. Concretely, using the atomicity of the file move (atomic move), even if it fails midway and re-runs, the final state converges uniquely. "The result doesn't change no matter how many times you repeat the same operation" — this is the key that makes large transfers over an unstable network safely retryable.

The hand of having the client hold the idempotency key: a content-hash works for "eliminating identical contents," but with a client-driven approach, combining an Idempotency-Key header (a unique ID the client numbers) lets you more clearly judge "a resend with the same intent." A common pattern in payment APIs, etc. Choose by requirement (YAGNI: for simple de-duplication, a content-hash alone suffices).


9. Testing: Verify multipart Uploads with TestClient

There's no production shipping without a verification path. FastAPI's TestClient (httpx-based) can reproduce a multipart upload with the files= argument. The boundaries (size limit, MIME spoofing, path traversal) are exactly what should be solidified with tests.

import io
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_upload_accepts_valid_png():
    # 正常系:本物の PNG マジックバイトで始まるバイト列
    png_bytes = b"\x89PNG\r\n\x1a\n" + b"0" * 100
    res = client.post(
        "/uploadfile/",
        files={"file": ("photo.png", io.BytesIO(png_bytes), "image/png")},
    )
    assert res.status_code == 200

def test_upload_rejects_mime_spoofing():
    # 異常系:Content-Type は image/png だが中身は PNG ではない(偽装)
    fake = b"MZ\x90\x00this-is-an-exe"     # 実行ファイルのシグネチャ
    res = client.post(
        "/uploadfile/",
        files={"file": ("photo.png", io.BytesIO(fake), "image/png")},
    )
    assert res.status_code == 415          # 第5章のマジックバイト検証で弾かれる

def test_upload_rejects_oversized_file():
    # 異常系:上限超過は 413
    big = io.BytesIO(b"\x89PNG\r\n\x1a\n" + b"0" * (MAX_FILE_SIZE + 1))
    res = client.post(
        "/uploadfile/",
        files={"file": ("big.png", big, "image/png")},
    )
    assert res.status_code == 413

def test_form_and_file_together():
    # Form と File の混在
    res = client.post(
        "/assets/",
        data={"meta_json": '{"title": "Intro", "tags": ["news"]}'},   # Form フィールド
        files={"file": ("clip.mp4", io.BytesIO(b"\x00\x00\x00\x18ftyp"), "video/mp4")},
    )
    assert res.status_code == 200

Beyond the normal case, always write (1) MIME spoofing, (2) size overflow, and (3) a filename aiming at path traversal as abnormal cases — here is where the gap between "works" and "safe" is filled. With only the former, you miss bugs in the validation wiring itself (a forgotten seek(0), a missed limit check).


10. Summary: A Production FastAPI File-Handling Cheat Sheet

A quick reference for when you're lost.

  • The first branch: over a few MB, consider presigned-URL direct. Protect the API's memory, bandwidth, and cost. Receive only small files via the API (hybrid is the realistic solution).
  • How to receive: as a rule UploadFile (spool = evacuate to disk above the memory threshold, async I/F, metadata). bytes (File) is limited to definitely small files. Reception needs python-multipart.
  • Large files: don't bulk-load with await file.read() (argument-less). Chunk-read with read(size), and enforce the size limit with a running byte count413. Keep Content-Length as an early-rejection hint, and re-validate with the actual byte count.
  • Filename: don't use UploadFile.filename in the save path. Use a server-numbered unique name, and confirm the resolved path is under the base directory (path-traversal defense).
  • Contents: three-stage validation with extension + content_type + leading magic bytes. After validation, rewind with await file.seek(0). Sort into clean/quarantine with malware scanning (decouple scanning from the request).
  • Forms: form values with Form(). Form and File can mix. But a JSON body can't be received in the same request (multipart). Receive structured data as a JSON string via Form and validate at the boundary.
  • The return side: dynamic generation is StreamingResponse (generator, backpressure), an existing file is FileResponse (ETag/Last-Modified auto).
  • Idempotency: make a content-hash the unique key for de-duplication. Place it with an atomic move so the write-in-progress isn't visible and it's idempotent against retries.
  • Testing: reproduce multipart with TestClient's files=. Always write MIME spoofing, size overflow, and path traversal as abnormal cases.

FastAPI is a framework where you can "get an upload working in 5 minutes," but production quality is decided by the design of the boundary. Stream the byte sequence in chunks, enforce the size while reading, have the server number the filename, validate the contents by substance, scan and quarantine external material, and make operations idempotent — none is flashy, but this accumulation makes file handling that "doesn't fall over, isn't contaminated, and is traceable even when a GiB-class big one comes."

I designed, in an in-house AI platform for a broadcaster, a zero-trust entrance that stream-scans up to GiB-class video/image material brought in from outside without exhausting memory, sorts malware into clean/quarantine with ClamAV, and is idempotent against retries by the atomicity of the move. With generative AI (Claude Code) as a partner, I build solo, fast, and cheap while guaranteeing quality with verification gates like boundary tests — that's my way of proceeding. For the design of the authentication/authorization boundary, read the FastAPI authentication/authorization guide, and for the operation of decoupling heavy post-processing from the request, the production-operations guide, alongside.

"I want to put file upload on this API in FastAPI, but how should I design receive-via-API vs presigned-URL-direct, large files, validation, and idempotency?" — from that judgment through implementation, verification, and operation, I accompany you end to end. Feel free to consult me even from the requirements-organizing stage.


References (Official Documentation)

友田

友田 陽大

Developer of a METI Minister's Award–winning product. With TypeScript + Python + AWS, I deliver SaaS, industry DX, and production-grade generative AI (RAG) end to end — from requirements to infrastructure and operations — single-handedly.

Got a challenge?

From design to implementation and operations — solo × generative AI

Implementation like this article's, end to end from requirements to production. Start with a free 30-minute technical consult and tell me about your situation.

Available for both project-based (contract) and advisory engagements. Start with a free 30-minute consult.

Also worth reading