大容量ファイルに強くなる：FastAPIのアップロード／ダウンロード設計――ストリーミング、レンジ対応、署名URL、ウイルススキャン、整合性検証
要約（最初に全体像）
- ファイルを「小物（～10MB）」「中物（～100MB）」「大物（GB級）」に分け、APIサーバで扱うか／外部ストレージへ直接流すかを切り替える。
- 非同期ストリーミングでメモリ使用量を一定に保ち、Content-Length／ETagで整合性を担保する。
- Range（部分取得）や圧縮・トランスコードはストレージやCDNに寄せる。APIはメタデータ管理と署名URLの発行に集中する。
- セキュリティは拡張子とMIMEの二重検証、ファイルサイズ上限、ウイルススキャン、保存前ハッシュで段階防御する。
誰が読んで得をするか
- 学習者Aさん（卒研で動画配信の原型を作りたい）
小さなAPIで安全に大きなファイルを扱う基礎を知りたい。 - 小規模チームBさん（受託・3名）
クライアントから大量の画像を一括アップロードさせたいが、サーバのメモリを食わせたくない。 - SaaS開発Cさん（スタートアップ）
S3互換ストレージを使い、署名URLによる直PUT／直GETを中心に据えたい。
1. 方針決め：サイズと経路で切り替える
1.1 3つのクラス分け
- 小物（～10MB）：APIサーバで受け取り→ストレージへ保存。バリデーションが簡単。
- 中物（～100MB）：APIサーバはストリーミングで受け渡し。CPU・メモリ／I/Oの監視必須。
- 大物（GB級）：署名URL（S3互換など）でクライアントが直接アップロード／ダウンロード。APIはメタデータと権限のみ。
1.2 役割分担
- API：認可、メタデータ、前後処理（ウイルススキャン、サムネ生成のキック）
- ストレージ／CDN：保存、レンジ配信、圧縮、キャッシュ
- ワーカー：重い変換やウイルススキャン
要点
- まずは経路設計。重いI/OはAPIから外すことを前提に。
2. 小物向け：Formデータでの基本アップロード
# app/main.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from pathlib import Path
import magic # python-magic 等、MIME推定用（導入は任意）
import hashlib
app = FastAPI(title="File Upload Basics")
BASE = Path("uploads")
BASE.mkdir(exist_ok=True)
MAX_MB = 10
def _digest_and_validate(file_bytes: bytes, expect_discrete: bool = True):
# MIME判定（拡張子だけに依存しない）
mime = magic.from_buffer(file_bytes[:4096], mime=True)
# 例：画像だけ許可
if not mime.startswith("image/"):
raise HTTPException(400, "unsupported media type")
# 内容ハッシュ（ETagのベースにできる）
etag = hashlib.sha256(file_bytes).hexdigest()
# 画像フォーマット偽装などの検査はここで拡張
return mime, etag
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
size_hint = 0
chunks = []
while True:
chunk = await file.read(1024 * 1024) # 1MB
if not chunk:
break
chunks.append(chunk)
size_hint += len(chunk)
if size_hint > MAX_MB * 1024 * 1024:
raise HTTPException(413, "file too large")
content = b"".join(chunks)
mime, etag = _digest_and_validate(content)
dest = BASE / file.filename
dest.write_bytes(content)
return {"filename": file.filename, "size": size_hint, "mime": mime, "etag": etag}
- 拡張子とMIMEの二重検証で偽装を軽減。
- 413 Payload Too Largeで早期に拒否。
- 小物なら一括読みも許容可。ただしメモリ使用量に注意。
要点
- 小さいうちは「まず正しく安全に」。MIME検査とサイズ制限が第一歩。
3. 中物向け：非同期ストリーミングでメモリ安定化
# app/stream.py
from fastapi import APIRouter, UploadFile, File, HTTPException
from pathlib import Path
import aiofiles
import hashlib
router = APIRouter(prefix="/stream", tags=["stream"])
CHUNK = 1024 * 1024 # 1MB
LIMIT = 100 * 1024 * 1024 # 100MB
DEST = Path("streams"); DEST.mkdir(exist_ok=True)
@router.post("/upload")
async def upload_stream(file: UploadFile = File(...)):
size = 0
sha256 = hashlib.sha256()
dest = DEST / file.filename
async with aiofiles.open(dest, "wb") as f:
while True:
chunk = await file.read(CHUNK)
if not chunk:
break
size += len(chunk)
if size > LIMIT:
raise HTTPException(413, "file too large")
sha256.update(chunk)
await f.write(chunk)
return {"filename": file.filename, "size": size, "etag": sha256.hexdigest()}
- aiofilesでノンブロッキング書き込み。
- ハッシュをストリーム計算し、整合性の基準（ETag）に使える。
- Webサーバ（Nginx等）の
client_max_body_sizeも併せて設定。
要点
- 逐次読み書きでピークメモリを一定に。ハッシュは並走して計算。
4. 大物向け：署名URLで「直接」アップロード／ダウンロード
4.1 なぜ署名URLか
- APIサーバを経由させないため、帯域とCPU負荷を回避。
- 一時的に有効なURL（数分）を払い出し、権限と有効期限で安全に。
4.2 流れ（アップロード）
- クライアント：
POST /files/initにファイル名・サイズ・MIMEを送る - API：ストレージ用の**署名URL（PUT）**を発行し返す
- クライアント：署名URLに直接PUT
- API：完了通知を受け、メタデータを確定（DB保存）
擬似コード（署名URL発行部分の例）：
# app/signed.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from datetime import timedelta
router = APIRouter(prefix="/files", tags=["signed"])
class InitReq(BaseModel):
filename: str = Field(..., max_length=256)
size: int = Field(..., ge=1)
mime: str
@router.post("/init")
async def init_upload(req: InitReq):
if req.size > 10 * 1024 * 1024 * 1024: # 10GB
raise HTTPException(413, "file too large")
# ここでMIMEや拡張子、禁止ワードなど細かいバリデーションを実施
# 署名URLの生成は利用ストレージのSDK等で実装（例：有効期限=5分）
signed_put_url = "<signed-put-url>"
object_key = f"user-uploads/{req.filename}"
return {"object_key": object_key, "put_url": signed_put_url, "expires_in": 300}
ダウンロードはGETの署名URLを払い出すだけ。Rangeや圧縮はストレージ／CDNが担当。
要点
- APIはURLを発行し、メタデータと認可に集中。大物は触らない。
5. ダウンロード：Range対応と条件付き応答
5.1 直接配信（小物・中物）
# app/download.py
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse, FileResponse, Response
from pathlib import Path
import os
router = APIRouter(prefix="/download", tags=["download"])
BASE = Path("streams")
@router.get("/file/{name}")
async def get_file(name: str):
path = BASE / name
if not path.exists():
raise HTTPException(404, "not found")
# 小物なら FileResponse でもOK
return FileResponse(path, media_type="application/octet-stream", filename=name)
5.2 Range（部分取得）
RangeはWebサーバ／ストレージに任せるのが基本だが、必要なら自実装も可能。大物では極力ストレージ直配信を選ぶ。
要点
- 直接配信は小物だけに。Rangeは**下流（Nginx/CDN/ストレージ）**を活用。
6. セキュリティ：段階防御チェックリスト
- サイズ上限：アプリ・Nginx・リバースプロキシで多層に設定。
- MIME＋拡張子の二重検査：偽装を軽減。
- 保存先の拡張子固定：ユーザー入力そのままを使わない（ディレクトリトラバーサル対策）。
- ウイルススキャン：クラウドAV／外部スキャンAPI／ワーカーで非同期化。
- ハッシュ検証：受領時のETag（sha256など）をメタデータに記録。
- 署名URLの短寿命化・パススコープ化：最小権限原則。
- ログと追跡：アップロード者・オブジェクトキー・ハッシュ・IP・時刻を記録。
要点
- 入力→保存→配信の各段に防御を置く。短寿命・最小権限が基本。
7. メタデータ管理と整合性
7.1 推奨スキーマ例
- id（内部ID）
- object_key（ストレージキー）
- filename（表示用）
- mime
- size
- etag（sha256等）
- owner_id（権限判定用）
- state（init / uploaded / verified / ready）
- created_at, updated_at
7.2 状態遷移
init（URL発行）→ クライアントがPUT →uploaded- ワーカーがウイルススキャン・ハッシュ検証 →
verified - トランスコード・サムネ生成 →
ready
要点
- 状態機械で管理。非同期処理でも追跡しやすい。
8. 非同期ジョブとの連携（重い処理は外で）
- ウイルススキャン、サムネ生成、動画のトランスコードはワーカーに。
- 失敗時は再試行（指数バックオフ）と手動再投入の仕組みを。
- 完了時はWebhook／通知でフロントへ反映。
要点
- APIは受付と通知に集中。ジョブで重い処理を吸収。
9. 監視と運用
- メトリクス：アップロード件数、失敗率、平均サイズ、署名URLの発行数、ウイルス検出件数。
- アラート：413の急増、署名URLの失効率、スキャン失敗の連続発生。
- ダッシュボード：期間ごとの容量推移、拡張子別分布、MIMEの分布。
要点
- 数値で「増えている／詰まっている」を把握。早期警戒が大切。
10. サンプル：まとめて動かす最小アプリ
# app/main.py
from fastapi import FastAPI
from app.stream import router as stream_router
from app.download import router as download_router
from app.signed import router as signed_router
app = FastAPI(title="File Handling Best Practices")
app.include_router(stream_router)
app.include_router(download_router)
app.include_router(signed_router)
@app.get("/health")
def health():
return {"ok": True}
11. よくある落とし穴と回避策
| 症状 | 原因 | 対策 |
|---|---|---|
| メモリが跳ね上がる | 一括読込 | ストリーミングへ切替、チャンクサイズ調整 |
| 偽装ファイルが混入 | 拡張子しか見ていない | MIME＋拡張子の二重検査、危険MIMEは拒否 |
| レンジ配信が重い | APIが配信している | CDN／ストレージ直配信、署名URLで権限制御 |
| URLが漏れて悪用 | 長寿命・広範囲の権限 | 短寿命・スコープ限定の署名URL |
| 事故時の追跡が困難 | ログ不足 | メタデータ＋操作ログを構造化して保存 |
12. 導入ロードマップ
- 小物：アップロードのサイズ上限とMIME検査、固定ディレクトリ保存。
- 中物：非同期ストリーミング、ハッシュ計算、FileResponseの適用。
- 大物：署名URLで直PUT／直GET、APIはメタデータ管理に特化。
- セキュリティ：AVスキャン、短寿命URL、最小権限。
- 運用：メトリクス・アラート・ダッシュボード整備。
13. 参考キーワード（検索の取っかかり）
- FastAPI UploadFile, StreamingResponse, FileResponse
- Content-Range / Range requests
- S3 pre-signed URL / 署名URL 発行
- python-magic MIME 判定
- ウイルススキャン API / clamav
- ETag / Content-Length 整合性
- Nginx client_max_body_size / proxy_read_timeout
- CDN Range / キャッシュ制御
まとめ
- ファイルはサイズと用途で経路を切り替える。小物はAPIで安全に、中物はストリーミング、大物は署名URLで直接。
- セキュリティはサイズ上限・二重検査・短寿命URL・AVスキャン・ハッシュの重ね掛け。
- 配信はストレージ／CDNに任せ、APIは認可・メタデータ・通知に特化する。これが速くて壊れにくい実装の近道。