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

> FastAPIのファイルアップロード（UploadFile/File）・フォーム（Form）・ストリーミング（StreamingResponse/FileResponse）を本番品質で扱うガイド。署名付きURL直行の設計判断、チャンク読みのメモリ枯渇回避、サイズ上限・MIME/マジックバイト検証・パストラバーサル防御・マルウェア検査・content-hash冪等まで実コードで解説します。

- 公開日: 2026-06-26
- 著者: 友田 陽大
- タグ: Python, FastAPI, ファイルアップロード, ストリーミング, セキュリティ
- URL: https://tomodahinata.com/blog/fastapi-file-uploads-form-data-streaming-responses-guide

## 要点

- 最初の分岐は『APIで受けるか、署名付きURLでオブジェクトストレージへ直行させるか』。数MB超や数GiB級はAPIのメモリ・帯域・コストを守るため presigned URL 直行が定石（YAGNI）
- UploadFile はスプール（メモリ閾値超でディスク退避）・async I/F・filename/content_type メタデータを持つ。File(...)でbytesは全部メモリに載るので小さいファイル限定。受信には python-multipart が必要
- 大容量は await file.read() で一括ロードせず、while で size 単位のチャンク読みにする。逐次バイトカウントでサイズ上限を強制し、Content-Length は信用しすぎない
- セキュリティは多層：拡張子＋Content-Type＋先頭マジックバイトを検証、ファイル名は自前生成（パストラバーサル防御）、マルウェア検査でクリーン/隔離に振り分け、content-hash をキーに冪等化
- 大きな生成結果やダウンロードは StreamingResponse（ジェネレータ）/FileResponse で逐次返す。Form と File は同一リクエストで混在できるが、同時に JSON ボディは受けられない（multipart）

---

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

この記事は、FastAPI で**本番品質のファイルアップロード・フォーム処理・ストリーミングレスポンス**を実装するためのガイドです。FastAPI 公式チュートリアルの `UploadFile` / `Form` / `StreamingResponse` を**最新の公式仕様に忠実に**追いながら、公式が（教材ゆえに）触れない領域——**大容量をメモリ枯渇させない受け方、サイズ上限の強制、MIME・マジックバイト検証、パストラバーサル防御、マルウェア検査、冪等化、そして「そもそもAPIで受けるべきか」という設計判断**——まで踏み込みます。題材として、私が国内大手放送事業者向けに構築した社内AIプラットフォーム（[複数のFastAPIサービスを束ねたモノレポ](/case-studies/broadcaster-ai-content-platform)。外部から持ち込まれる**最大数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`** が要ります（アップロードファイルは「フォームデータ」として送られるため）。

```bash
pip install python-multipart
```

```python
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` を忘れないこと。

```python
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` 許容にします。

```python
@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]` にするだけです。

```python
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：全部メモリに載せる

```python
# ❌ アンチパターン：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：チャンクで読み、逐次バイトカウントで上限を強制する

```python
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）こともできます。

```python
# 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 に届く前に巨大ボディを遮断するのが本番の多層防御です。重い後処理（変換・解析）をリクエストから切り離す設計は[本番運用ガイド](/blog/fastapi-production-async-pydantic-observability-guide)を併読してください。

---

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

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

**最も安全なのは、クライアント指定のファイル名を保存パスに一切使わないこと**です。サーバー側で一意な名前を生成し、元のファイル名は（必要なら）メタデータとして DB に持つだけにします。

```python
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 → マジックバイト

```python
# 各形式の「マジックバイト（ファイルシグネチャ）」。中身の先頭で本物か判定する。
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 ハンドラ内で同期的に走らせるのは禁物——**アップロードは一旦『未検査』ゾーンに置き、検査は非同期ジョブ／イベント駆動**で行い、結果でクリーン/隔離へ移します。

```python
# 概念図：アップロード直後は"検査前ゾーン"。検査はリクエストから切り離す。
# 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` 単体

```python
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 で混在**できます。

```python
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 に流し込む**（境界で検証）。

```python
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チャンク分だけです。

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

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

```python
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` を**自動付与**します。

```python
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** することもできます。

```python
@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回目以降は**保存をスキップして既存の参照を返す**。

```python
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 偽装・パストラバーサル）こそ、テストで固めるべき箇所です。

```python
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 認証・認可ガイド](/blog/fastapi-authentication-oauth2-jwt-security-scopes-production-guide)、重い後処理をリクエストから切り離す運用は[本番運用ガイド](/blog/fastapi-production-async-pydantic-observability-guide)を併読してください。

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

---

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

- [Request Files（FastAPI）](https://fastapi.tiangolo.com/tutorial/request-files/) — `File`・`UploadFile`・`bytes` との違い・スプール・`list[UploadFile]`・`python-multipart`
- [Request Forms（FastAPI）](https://fastapi.tiangolo.com/tutorial/request-forms/) — `Form` の宣言・エンコーディング・JSON ボディと併用できない制約
- [Request Forms and Files（FastAPI）](https://fastapi.tiangolo.com/tutorial/request-forms-and-files/) — `Form` と `File` の混在・multipart の制約
- [Custom Response（FastAPI）](https://fastapi.tiangolo.com/advanced/custom-response/) — `StreamingResponse`・`FileResponse`・`Response`・`media_type`
- [UploadFile（FastAPI リファレンス）](https://fastapi.tiangolo.com/reference/uploadfile/) — `filename`/`content_type`/`file` と `read`/`write`/`seek`/`close`
- [python-multipart](https://github.com/Kludex/python-multipart) — multipart/form-data のパーサ（ファイル/フォーム受信の前提）