FastAPI ファイルアップロード・フォーム・ストリーミング本番ガイド：UploadFile / Form / StreamingResponse をメモリ枯渇させず安全・冪等に

「FastAPI でファイルアップロードを付けたい」——要件は一行です。けれどファイル受信は、一行のミスがそのままサーバーのメモリ枯渇・ディスク破壊・任意ファイル書き込みになる領域です。await file.read() で 5GiB の動画を丸ごとメモリに載せてプロセスが落ちた、アップロードされた filename をそのままパスに使って ../../etc/passwd を書かれた、Content-Type を信じて実行ファイルを画像として保存した——どれも「小さいファイルなら動いてしまう」がゆえに、本番でユーザーが大物を投げてくるまで気づきません。

この記事は、FastAPI で本番品質のファイルアップロード・フォーム処理・ストリーミングレスポンスを実装するためのガイドです。FastAPI 公式チュートリアルの UploadFile / Form / StreamingResponse を最新の公式仕様に忠実に追いながら、公式が（教材ゆえに）触れない領域——大容量をメモリ枯渇させない受け方、サイズ上限の強制、MIME・マジックバイト検証、パストラバーサル防御、マルウェア検査、冪等化、そして「そもそもAPIで受けるべきか」という設計判断——まで踏み込みます。題材として、私が国内大手放送事業者向けに構築した社内AIプラットフォーム（複数のFastAPIサービスを束ねたモノレポ。外部から持ち込まれる最大数GiBの動画・画像素材を、プラットフォーム到達前にマルウェア検査し、クリーン/隔離に振り分けるゼロトラストな入口を設計）での判断も交えます。

この記事のルール：API・推奨ライブラリは FastAPI 公式ドキュメント（2026年6月時点） に基づきます。ファイル/フォームの受信には python-multipart が必須です（公式が明記）。仕様は改定されるため、本番投入前に必ず公式で最新の挙動を確認してください。シークレット（バケット名・署名鍵・接続情報）は環境変数前提（ハードコード厳禁）。外部から届くファイルはすべて信用しない——これが本記事の通底するルールです。

0. まず判断：APIで受けるか、署名付きURLで直行させるか

実装に入る前に、本番設計で最初に決めるべき分岐があります。「そのファイルを、本当に FastAPI プロセスを通して受ける必要があるか？」 です。ここを飛ばして全部 API で受けると、アップロードのたびに API のメモリ・帯域・CPU・コストが食われ、スケールの足枷になります。

選択肢は2つです。

(A) API で受ける：クライアント → FastAPI → ストレージ。FastAPI がバイト列を中継する。検証・変換をその場でやりやすい。
(B) 署名付きURL（presigned URL）で直行：FastAPI は「アップロード先の署名付きURL」を発行するだけ。クライアント → オブジェクトストレージ（S3 / GCS）へ直接 PUT する。バイト列は API を通らない。

観点	(A) APIで受ける	(B) 署名付きURLで直行
向くケース	小さいファイル（〜数MB）、その場で検証/変換したい	大容量（数十MB〜数GiB）、画像/動画/バックアップ
API のメモリ・帯域	全バイトが API を通る（負荷源）	API を通らない（プロセスは軽いまま）
スケール	API のスループットが上限	ストレージ側がスケール（API は無関係）
コスト	API の実行時間・転送が嵩む	API はURL発行のみ＝安い
検証のタイミング	受信時にその場で	アップロード完了後にイベント駆動で（後追い）
実装の単純さ	単純（KISS）	署名・CORS・完了通知の配線が要る

判断の目安：数MBを超える可能性があるなら、まず (B) 署名付きURL直行を検討してください。「とりあえず全部 API で受ける」は、小さいうちは動きますが、ユーザーが大物を投げた瞬間にメモリ枯渇でプロセスごと落ちます。本記事は (A) を正しく安全にやる方法を中心に解説しつつ、(B) の設計（第7章）も具体的に示します。多くの本番系は「小さいファイルは A、大容量は B」のハイブリッドが現実解です。

放送事業者向けプラットフォームでは、最大数GiBの素材アップロードを**署名付きURLのチャンク並列（最大8並列）**でストレージへ直行させ、API プロセスにバイト列の中継をさせませんでした。これが「数GiB級でもメモリを枯渇させない」入口設計の核です。

1. `UploadFile` と `bytes`（`File`）：どちらで受けるか

API で受ける（選択肢 A）と決めたら、最初の判断は UploadFile で受けるか、bytes（File）で受けるかです。結論から言うと、本番は原則 UploadFile です。理由は明確です。

まず両者の最小コードを、公式どおりに並べます。受信には python-multipart が要ります（アップロードファイルは「フォームデータ」として送られるため）。

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 `UploadFile` の利点（公式が挙げる4点）

公式は UploadFile が bytes より優れている理由を、はっきり列挙しています。

File() をデフォルト値に書かなくてよい（型注釈だけで成立）。
「スプールされた（spooled）」ファイルを使う——一定サイズまではメモリに保持し、その閾値を超えるとディスクに退避する。
これにより、画像・動画・大きなバイナリでも、メモリを食い尽くさずに扱える。
アップロードファイルからメタデータを取得できる。
ファイルライクな async インターフェースを持つ。

対して bytes（File）は、ファイルの中身を丸ごとメモリに読み込みます。len(file) がすぐ取れて手軽ですが、大容量では即メモリ枯渇します。だから bytes は「数KB〜せいぜい数百KBの、確実に小さいと分かっているファイル」に限定すべきです。

要点：bytes の手軽さは「小さいと保証できる場合だけ」の手軽さです。アップロードサイズはクライアントが決めるものであり、あなたの想定を超える物が必ず来ます。迷ったら UploadFile——スプールがメモリ枯渇を構造的に防ぎます。

1.2 `UploadFile` のメタデータと async メソッド

UploadFile は次のメタデータを持ちます。

filename：アップロード元のファイル名（例 myimage.jpg）。信用してはいけない値（第4章でサニタイズ）。
content_type：MIME タイプ（例 image/jpeg）。これも自己申告で信用しすぎない（第5章で検証）。
file：SpooledTemporaryFile（ファイルライクオブジェクト）。

そして、操作は async メソッドです。await を忘れないこと。

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

await file.read() を無条件で呼ばない：引数なしの await file.read() は全バイトをメモリに展開します。これは bytes で受けるのと同じメモリ消費——UploadFile のスプールの恩恵を自分で捨てる行為です。大容量を想定するなら、read(size) でチャンク読みします（第3章）。「UploadFile を使っているから安全」ではなく「チャンクで読むから安全」です。

2. 単一ファイルと複数ファイル

2.1 任意（オプショナル）アップロード

ファイルを必須にしないなら、公式どおり None 許容にします。

@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 複数ファイルは `list[UploadFile]`

複数ファイルを受けるなら、型を 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]}

複数ファイルでも上限は「総量」で考える：1ファイルのサイズ上限だけ守っても、10,000 個の小ファイルを一度に送られればメモリもディスクも食い潰されます。本番では 「1ファイルの上限」と「1リクエストの総バイト数・ファイル数の上限」の両方を強制してください（第3章）。

3. 大容量を安全に：チャンク読み・サイズ上限・バックプレッシャ

ここが本章の心臓です。「動く」と「大容量でも落ちない」を分けるのは、ほぼこの一点——バイト列を一括ではなくチャンク（小片）でストリーム処理することです。

3.1 NG：全部メモリに載せる

# ❌ アンチパターン：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)}

これは二重に危険です。(1) 全バイトをメモリに載せる。(2) ユーザー指定の filename をそのまま書き込みパスに使う（パストラバーサル）。

3.2 OK：チャンクで読み、逐次バイトカウントで上限を強制する

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

ポイントは3つです。

常時メモリに載るのは1チャンク（ここでは1MiB）だけ。ファイルが5GiBでもメモリ消費は一定。これがバックプレッシャ（読んだ分だけ処理して捨てる）の基本形です。
サイズ上限は「逐次バイトカウント」で強制する。全部受け取ってから len() で測るのでは、その時点で既にメモリ/ディスクを食われています。読みながら超えた瞬間に中断するのが正解。
上限を超えたら 413 Request Entity Too Large を返す。

3.3 `Content-Length` を信用しすぎない

「Content-Length ヘッダを見て、大きすぎたら弾けばいい」と思いがちですが、Content-Length は自己申告です。攻撃者は小さい値を申告して大量に送りつけることも、ヘッダを付けない（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 の逐次カウントで必ず再検証する。

多層で守る：Content-Length での早期拒否は「無駄な転送を早く切る」ための最適化に過ぎません。**最終的な真実は「実際に読んだバイト数」**です。さらに前段の リバースプロキシ（Nginx の client_max_body_size）やロードバランサ・WAF でもボディサイズ上限をかけ、API に届く前に巨大ボディを遮断するのが本番の多層防御です。重い後処理（変換・解析）をリクエストから切り離す設計は本番運用ガイドを併読してください。

4. ファイル名のサニタイズ：パストラバーサルを防ぐ

UploadFile.filename は、外部から来る信用できない文字列です。これを保存パスにそのまま使うと、../../etc/cron.d/evil のようなパストラバーサルで、意図しない場所にファイルを書き込まれます。

最も安全なのは、クライアント指定のファイル名を保存パスに一切使わないことです。サーバー側で一意な名前を生成し、元のファイル名は（必要なら）メタデータとして 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

「サニタイズして使う」より「使わない」：filename から / や .. を除去する“サニタイズ”もありますが、エンコーディングの抜け穴（%2e%2e、NUL バイト、Unicode 正規化）で破られがちです。サーバーが採番した一意名を使い、元の名前は表示用メタデータに留めるのが、最も堅牢で単純（KISS）です。放送事業者向けプラットフォームでも、素材は Cloud Storage に保管しつつパストラバーサル防御を入口に組み込み、外部由来の名前を保存パスに直結させない設計にしました。

5. 中身を検証する：拡張子・MIME・マジックバイト・マルウェア検査

ファイル名と Content-Type は自己申告です。「.jpg だから画像」「Content-Type: image/png だから安全」は成り立ちません。攻撃者は実行ファイルを photo.jpg と名乗らせ、image/png を申告できます。実体（中身の先頭バイト＝マジックバイト）で検証します。

5.1 三段の検証：拡張子 → Content-Type → マジックバイト

# 各形式の「マジックバイト（ファイルシグネチャ）」。中身の先頭で本物か判定する。
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

ここで効くのが await file.seek(0) です。検証で先頭を読んだら、本保存の前に必ず巻き戻す——でないと、保存されるファイルが先頭8バイト欠けます。

マジックバイトは「最低限」であって万能ではない：先頭シグネチャの一致は偽装の大半を弾きますが、ポリグロット（複数形式として妥当なファイル）や、形式は正しいが悪性なファイル（細工された PDF・画像内の埋め込みスクリプト）は通します。本番ではマジックバイト検証＋マルウェア検査を重ねます。python-magic（libmagic バインディング）でより堅牢に MIME 判定する手もありますが、外部ライブラリ追加の是非は要件次第（YAGNI）。

5.2 マルウェア検査とクリーン/隔離の振り分け

外部から持ち込まれるファイルは、保存・公開する前にマルウェア検査し、クリーン/隔離（quarantine）に振り分けるのが本番の定石です。検査エンジン（ClamAV など）は重くて遅いため、API ハンドラ内で同期的に走らせるのは禁物——アップロードは一旦『未検査』ゾーンに置き、検査は非同期ジョブ／イベント駆動で行い、結果でクリーン/隔離へ移します。

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

放送事業者向けプラットフォームでは、外部から持ち込まれる動画・画像素材をプラットフォーム到達前に ClamAV（Cloud Run）でストリーミング検査し、バッファせずにメモリ枯渇を回避しながら、クリーン/隔離バケットへ振り分けました。検査・移動は最大数GiBの素材でも完走し、移動の原子性によって再試行に対し冪等になるよう設計しています。「検査前の素材は決してプラットフォーム内部に到達させない」——このゼロトラストな入口が、外部素材を扱う AI 基盤の前提でした。

6. フォームデータ（`Form`）と `File` の混在

ファイルと一緒に**メタデータ（タイトル・カテゴリ・トークン）**を送りたいことは多い。このとき使うのが Form です。

6.1 `Form` 単体

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}

Form を明示しないと、その引数はクエリパラメータか JSON ボディと解釈されます。フォーム値として受けるには Form() が必須です（Form は Body を直接継承するクラス）。フォーム受信にも python-multipart が要ります。

6.2 `Form` と `File` を同じリクエストで受ける

公式どおり、File と Form は同一の 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 【重要】`Form`/`File` と JSON ボディは同時に受けられない

ここは公式が明確に警告する本番で必ず踏む落とし穴です。

公式の警告：1つの path operation で複数の Form / File を宣言できますが、JSON として受け取る Body フィールドを同時に宣言することはできません。リクエストボディが application/json ではなく multipart/form-data（または application/x-www-form-urlencoded）でエンコードされるためです。これは FastAPI の制約ではなく HTTP プロトコルの仕様です。

つまり、「ファイル＋構造化された JSON メタデータ」を1リクエストで送りたい場合、Pydantic モデルを Body として受けることはできません。実務での対処は次のどれかです。

メタデータも Form のフィールドとして平坦に送る（title: Annotated[str, Form()] など）。ネストが浅いならこれで十分。
構造化データは JSON 文字列を Form で受け、ハンドラ内で Pydantic に流し込む（境界で検証）。

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}

設計の含意：この HTTP の制約こそ、第0章で『大容量は署名付きURL直行』を勧める理由でもあります。(B) なら「メタデータ作成 API（JSON）」と「ファイル本体（ストレージへ直行）」を別リクエストに分離できるので、multipart の制約に縛られず、API は綺麗な JSON だけを扱えます。

7. ストリーミングレスポンス：大きな結果・ダウンロードを逐次返す

ここまでは「受け取る」側でした。「返す」側でも、大きなデータを一括でメモリに作ってから返すとメモリ枯渇します。FastAPI は逐次返すための道具を用意しています。

7.1 `StreamingResponse`：ジェネレータで逐次送る

StreamingResponse は、（async）ジェネレータが yield するバイト列を、生成しながら送出します。巨大なレスポンスでも、常時メモリに載るのは1チャンク分だけです。

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())

実用例として、ストレージ上の大きなファイルをチャンク読みしながらそのまま流す（プロキシ的ダウンロード）形を示します。

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"'},
    )

バックプレッシャとキャンセル：StreamingResponse の真価は、クライアントが受け取れる速度に合わせて生成が進むこと（バックプレッシャ）です。クライアントが切断したら生成を止められるよう、async ジェネレータ内で適切に await（anyio.sleep(0) など）を入れてキャンセルを協調的に扱えるようにします。LLM の逐次生成やログのテーリングなど、「全部できてから返す」では遅すぎる場面でも有効です。

7.2 `FileResponse`：ディスク上のファイルを返す定石

「サーバー上の既存ファイルをそのまま返す」だけなら、自分でジェネレータを書くより FileResponse が簡潔で安全です。非同期でファイルを読み、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",
    )

response_class=FileResponse を使えば、path operation からファイルパスを直接 return することもできます。

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

StreamingResponse と FileResponse の使い分け：ディスク上に既にあるファイルを返すなら FileResponse（ETag・Last-Modified まで面倒を見てくれる）。動的に生成する／外部から逐次取得するバイト列なら StreamingResponse。任意の media_type の小さな生データを返すだけなら、Response(content=..., media_type=...) で十分です（オーバーエンジニアリングしない）。

8. 冪等性：content-hash で重複アップロードを排除する

アップロードはネットワーク越しの操作であり、リトライがつきものです。クライアントがタイムアウトして再送し、同じファイルが2回保存される——これは事故です。本番では**冪等（idempotent）**に作ります。

定石は ファイルの内容ハッシュ（content-hash）を一意キーにすることです。同じ中身なら同じハッシュ＝同じオブジェクトとして扱い、2回目以降は保存をスキップして既存の参照を返す。

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}

放送事業者向けプラットフォームでも、長時間ジョブと素材移動を再試行に対して冪等に設計しました。具体的には、ファイル移動の原子性（atomic move）を使って、途中で失敗・再実行しても最終状態が一意に収束するようにしています。「同じ操作を何度繰り返しても結果が変わらない」——これが、不安定なネットワーク越しの大容量転送を安全に再試行可能にする鍵です。

冪等キーをクライアントに持たせる手も：content-hash は「中身が同じものの排除」に効きますが、クライアント主導なら Idempotency-Key ヘッダ（クライアントが採番する一意ID）を併用すると、「同じ意図の再送」をより明確に判定できます。決済 API などで一般的なパターンです。要件に応じて選んでください（YAGNI：単純な重複排除なら content-hash だけで十分）。

9. テスト：multipart アップロードを `TestClient` で検証する

検証パスのない本番投入はありません。FastAPI の TestClient（httpx ベース）は、files= 引数で multipart アップロードを再現できます。境界（サイズ上限・MIME 偽装・パストラバーサル）こそ、テストで固めるべき箇所です。

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

正常系だけでなく、(1) MIME 偽装、(2) サイズ超過、(3) パストラバーサルを狙う filename、を必ず異常系として書く——ここが「動く」と「安全」の差を埋めます。前者だけだと、検証配線そのもののバグ（seek(0) 忘れ、上限チェック漏れ）を見逃します。

10. まとめ：本番 FastAPI ファイル処理チートシート

迷ったときの早見表です。

最初の分岐：数MB超は 署名付きURL直行（presigned URL） を検討。API のメモリ・帯域・コストを守る。小さいファイルだけ API で受ける（ハイブリッドが現実解）。
受け方：原則 UploadFile（スプール＝メモリ閾値超でディスク退避・async I/F・メタデータ）。bytes（File）は確実に小さいファイル限定。受信に python-multipart 必須。
大容量：await file.read()（引数なし）で一括ロードしない。read(size) でチャンク読みし、逐次バイトカウントでサイズ上限を強制して 413。Content-Length は早期拒否のヒントに留め、実バイト数で再検証。
ファイル名：UploadFile.filename を保存パスに使わない。サーバーが採番した一意名を使い、解決後パスが基準ディレクトリ配下か確認（パストラバーサル防御）。
中身：拡張子＋content_type＋先頭マジックバイトで三段検証。検証後は await file.seek(0) で巻き戻す。マルウェア検査でクリーン/隔離に振り分け（検査はリクエストから切り離す）。
フォーム：Form() でフォーム値。Form と File は混在可。ただし同一リクエストで JSON ボディは受けられない（multipart）。構造化データは JSON 文字列を Form で受けて境界で検証。
返す側：動的生成は StreamingResponse（ジェネレータ・バックプレッシャ）、既存ファイルは FileResponse（ETag/Last-Modified 自動）。
冪等性：content-hash を一意キーに重複排除。配置は原子的な moveで書きかけを見せない・再試行に冪等。
テスト：TestClient の files= で multipart を再現。MIME 偽装・サイズ超過・パストラバーサルを異常系として必ず書く。

FastAPI は「5分でアップロードを動かせる」フレームワークですが、本番品質は境界の設計で決まります。バイト列をチャンクで流し、サイズを読みながら強制し、ファイル名はサーバーが採番し、中身を実体で検証し、外部素材は検査して隔離し、操作を冪等にする——どれも派手ではありませんが、この積み重ねが「数GiBの大物が来ても落ちない・汚染されない・追える」ファイル処理を作ります。

私は放送事業者向けの社内AIプラットフォームで、外部から持ち込まれる最大数GiB級の動画・画像素材を、メモリを枯渇させずストリーミング検査し、ClamAV でマルウェアをクリーン/隔離に振り分け、移動の原子性で再試行に冪等な入口を設計しました。生成AI（Claude Code）を相棒に、一人で速く・安く作りつつ、境界テストのような検証ゲートで品質を担保するのが私の進め方です。認証・認可の境界設計はFastAPI 認証・認可ガイド、重い後処理をリクエストから切り離す運用は本番運用ガイドを併読してください。

「FastAPI でこの API にファイルアップロードを載せたいが、APIで受けるか署名付きURL直行か、大容量・検証・冪等をどう設計すべきか」——その判断から実装・検証・運用まで、一気通貫で伴走します。要件整理の段階からでも、お気軽にご相談ください。

参考（公式ドキュメント）

Request Files（FastAPI） — File・UploadFile・bytes との違い・スプール・list[UploadFile]・python-multipart
Request Forms（FastAPI） — Form の宣言・エンコーディング・JSON ボディと併用できない制約
Request Forms and Files（FastAPI） — Form と File の混在・multipart の制約
Custom Response（FastAPI） — StreamingResponse・FileResponse・Response・media_type
UploadFile（FastAPI リファレンス） — filename/content_type/file と read/write/seek/close
python-multipart — multipart/form-data のパーサ（ファイル/フォーム受信の前提）

FastAPI ファイルアップロード・フォーム・ストリーミング本番ガイド：UploadFile / Form / StreamingResponse をメモリ枯渇させず安全・冪等に

0. まず判断：APIで受けるか、署名付きURLで直行させるか

1. `UploadFile` と `bytes`（`File`）：どちらで受けるか

1.1 `UploadFile` の利点（公式が挙げる4点）

1.2 `UploadFile` のメタデータと async メソッド

2. 単一ファイルと複数ファイル

2.1 任意（オプショナル）アップロード

2.2 複数ファイルは `list[UploadFile]`

3. 大容量を安全に：チャンク読み・サイズ上限・バックプレッシャ

3.1 NG：全部メモリに載せる

3.2 OK：チャンクで読み、逐次バイトカウントで上限を強制する

3.3 `Content-Length` を信用しすぎない

4. ファイル名のサニタイズ：パストラバーサルを防ぐ

5. 中身を検証する：拡張子・MIME・マジックバイト・マルウェア検査

5.1 三段の検証：拡張子 → Content-Type → マジックバイト

5.2 マルウェア検査とクリーン/隔離の振り分け

6. フォームデータ（`Form`）と `File` の混在

6.1 `Form` 単体

6.2 `Form` と `File` を同じリクエストで受ける

6.3 【重要】`Form`/`File` と JSON ボディは同時に受けられない

7. ストリーミングレスポンス：大きな結果・ダウンロードを逐次返す

7.1 `StreamingResponse`：ジェネレータで逐次送る

7.2 `FileResponse`：ディスク上のファイルを返す定石

8. 冪等性：content-hash で重複アップロードを排除する

9. テスト：multipart アップロードを `TestClient` で検証する

10. まとめ：本番 FastAPI ファイル処理チートシート

参考（公式ドキュメント）

最短ルート：カレンダーから直接予約

FastAPI 本番運用ガイド：async の正しい使いどころ・Pydantic v2 境界バリデーション・DIと可観測性で落ちないAPIを作る

Python のデータ型完全ガイド：数値・文字列・コレクションの『正しい使い分け』と本番で壊れない設計

Python のマッピング完全ガイド：dict の内部・collections の使い分け・自作マッピング設計と本番運用

FastAPI 認証・認可本番実装ガイド：OAuth2パスワードフロー × JWT（PyJWT）× Security scopesで守るAPI

Cũng đáng đọc

Pydantic エラーハンドリング実践ガイド：ValidationErrorを使いやすい・安全なAPIエラーに変える

JWTの署名はなぜ破れないのか — 暗号理論とPyJWT 2.13.0のソースコードで解剖する署名検証

Flask の認証実装ガイド：Flask-Login（セッション認証）と Flask-JWT-Extended（トークン認証）の使い分けと本番実装

0. まず判断：APIで受けるか、署名付きURLで直行させるか

1. UploadFile と bytes（File）：どちらで受けるか

1.1 UploadFile の利点（公式が挙げる4点）

1.2 UploadFile のメタデータと async メソッド

2. 単一ファイルと複数ファイル

2.1 任意（オプショナル）アップロード

2.2 複数ファイルは list[UploadFile]

3. 大容量を安全に：チャンク読み・サイズ上限・バックプレッシャ

3.1 NG：全部メモリに載せる

3.2 OK：チャンクで読み、逐次バイトカウントで上限を強制する

3.3 Content-Length を信用しすぎない

4. ファイル名のサニタイズ：パストラバーサルを防ぐ

5. 中身を検証する：拡張子・MIME・マジックバイト・マルウェア検査

5.1 三段の検証：拡張子 → Content-Type → マジックバイト

5.2 マルウェア検査とクリーン/隔離の振り分け

6. フォームデータ（Form）と File の混在

6.1 Form 単体

6.2 Form と File を同じリクエストで受ける

6.3 【重要】Form/File と JSON ボディは同時に受けられない

7. ストリーミングレスポンス：大きな結果・ダウンロードを逐次返す

7.1 StreamingResponse：ジェネレータで逐次送る

7.2 FileResponse：ディスク上のファイルを返す定石

8. 冪等性：content-hash で重複アップロードを排除する

9. テスト：multipart アップロードを TestClient で検証する

10. まとめ：本番 FastAPI ファイル処理チートシート

参考（公式ドキュメント）

最短ルート：カレンダーから直接予約

Bài viết liên quan

FastAPI 本番運用ガイド：async の正しい使いどころ・Pydantic v2 境界バリデーション・DIと可観測性で落ちないAPIを作る

Python のデータ型 完全ガイド：数値・文字列・コレクションの『正しい使い分け』と本番で壊れない設計

Python のマッピング完全ガイド：dict の内部・collections の使い分け・自作マッピング設計と本番運用

FastAPI 認証・認可 本番実装ガイド：OAuth2パスワードフロー × JWT（PyJWT）× Security scopesで守るAPI

Cũng đáng đọc

Pydantic エラーハンドリング実践ガイド：ValidationErrorを使いやすい・安全なAPIエラーに変える

JWTの署名はなぜ破れないのか — 暗号理論とPyJWT 2.13.0のソースコードで解剖する署名検証

Flask の認証実装ガイド：Flask-Login（セッション認証）と Flask-JWT-Extended（トークン認証）の使い分けと本番実装

1. `UploadFile` と `bytes`（`File`）：どちらで受けるか

1.1 `UploadFile` の利点（公式が挙げる4点）

1.2 `UploadFile` のメタデータと async メソッド

2.2 複数ファイルは `list[UploadFile]`

3.3 `Content-Length` を信用しすぎない

6. フォームデータ（`Form`）と `File` の混在

6.1 `Form` 単体

6.2 `Form` と `File` を同じリクエストで受ける

6.3 【重要】`Form`/`File` と JSON ボディは同時に受けられない

7.1 `StreamingResponse`：ジェネレータで逐次送る

7.2 `FileResponse`：ディスク上のファイルを返す定石

9. テスト：multipart アップロードを `TestClient` で検証する

Python のデータ型完全ガイド：数値・文字列・コレクションの『正しい使い分け』と本番で壊れない設計

FastAPI 認証・認可本番実装ガイド：OAuth2パスワードフロー × JWT（PyJWT）× Security scopesで守るAPI