FastAPIで大容量ファイルを扱う設計：アップロード、Range配信、署名URL、整合性検証

FastAPIで大容量ファイルを扱うときは、すべてのデータをAPIサーバーに通す前提を置かないことが出発点です。
小さなファイルはAPIで受け取り、大きなファイルはオブジェクトストレージへ直接送ると、認可、検査、保存、配信の役割を分けやすくなります。

この記事では、API経由のアップロード、署名URLによる直接転送、Rangeを使った部分取得、ウイルススキャン、ハッシュ検証を一つの処理フローとして整理します。
FastAPIそのものの特徴から確認したい場合は、FastAPIの特徴とAPI設計の基本も参考になります。

最初に決めるファイルの経路

ファイルサイズだけで実装を決めると、利用者数や同時実行数が増えたときに設計が崩れます。
サイズ、同時アップロード数、再開の必要性、検査内容、配信頻度を合わせて、ファイルが通る経路を選びます。

原稿で使っていた10MB、100MB、GB級という区切りは、設計を考えるための例です。
FastAPIが定める固定値ではないため、実際の上限はリバースプロキシ、アプリケーション、ストレージ、回線、運用時間の条件から決めます。

役割を分ける

• API：利用者の認証と認可、ファイル情報の登録、署名URLの発行、状態確認を担当します。
• ストレージとCDN：ファイルの保存、キャッシュ、Range配信を担当します。
• ワーカー：ウイルススキャン、サムネイル生成、動画変換など、時間のかかる処理を担当します。

ストリーミングとは、ファイル全体を一度にメモリへ載せず、小さな単位で順番に読み書きする方法です。
ただし、FastAPIのUploadFileは受信データをメモリまたは一時ファイルへスプールしてから扱うため、ハンドラー内のチャンク処理だけで通信経路全体の負荷が解決するわけではありません。

API経由でアップロードする実装

bytes型で受け取ると、ファイル全体をメモリに保持します。
一方、UploadFileは一定量を超えたデータを一時ファイルへ退避し、非同期のread()を提供するため、ファイルを分割して処理できます。

次の例は、100MBを上限として1MBずつ隔離領域へ保存し、同時にSHA-256を計算します。
上限超過や検査失敗が起きた場合は、途中まで保存したファイルを削除します。

from pathlib import Path
from uuid import uuid4
import hashlib

import aiofiles
import magic
from fastapi import APIRouter, HTTPException, UploadFile

router = APIRouter(prefix="/files", tags=["files"])

CHUNK_BYTES = 1024 * 1024
MAX_BYTES = 100 * 1024 * 1024
QUARANTINE = Path("quarantine")
QUARANTINE.mkdir(exist_ok=True)

ALLOWED_MIME = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
}


@router.post("/upload")
async def upload(file: UploadFile):
    declared_mime = file.content_type or ""
    if declared_mime not in ALLOWED_MIME:
        raise HTTPException(415, "unsupported media type")

    stored_name = f"{uuid4().hex}{ALLOWED_MIME[declared_mime]}"
    destination = QUARANTINE / stored_name
    digest = hashlib.sha256()
    size = 0

    try:
        async with aiofiles.open(destination, "xb") as output:
            while chunk := await file.read(CHUNK_BYTES):
                size += len(chunk)
                if size > MAX_BYTES:
                    raise HTTPException(413, "file too large")
                digest.update(chunk)
                await output.write(chunk)

        detected_mime = magic.from_file(str(destination), mime=True)
        if detected_mime not in ALLOWED_MIME:
            raise HTTPException(415, "file content is not allowed")

        return {
            "display_name": file.filename,
            "stored_name": stored_name,
            "size": size,
            "mime": detected_mime,
            "sha256": digest.hexdigest(),
        }
    except Exception:
        destination.unlink(missing_ok=True)
        raise
    finally:
        await file.close()

このコードでも、クライアントが送るContent-Typeは事前選別にしか使っていません。
保存後にファイルの内容から形式を推定し、公開領域ではない隔離先で検査します。

保存名には利用者が送ったfilenameを使わず、サーバー側で生成した値を使います。
元のファイル名は表示用メタデータとして保持すれば、../などを含む名前による意図しないパス参照と、同名ファイルの衝突を避けられます。

なお、MIME推定だけでファイルの安全性を保証することはできません。
許可する形式を絞り、必要に応じて対象形式のデコーダーによる検査とウイルススキャンを追加します。

署名URLでストレージへ直接送る

署名URLは、特定のオブジェクトとHTTPメソッドに対する一時的なアクセス権をURLとして渡す仕組みです。
クライアントがAPIサーバーを経由せずにストレージへPUTできるため、APIは大容量データの中継ではなく、認可と状態管理に集中できます。

1. クライアントがファイル名、予定サイズ、MIMEタイプをAPIへ送ります。
2. APIが利用者を認可し、推測しにくいオブジェクトキーと短い有効期限のPUT用URLを発行します。
3. クライアントが隔離用の保存領域へ直接アップロードします。
4. クライアントが完了をAPIへ通知します。
5. APIまたはワーカーが実際のサイズ、チェックサム、ファイル形式、スキャン結果を確認します。
6. 検査に合格したファイルだけをreadyへ進め、ダウンロードを許可します。

最初に申告されたサイズやMIMEタイプだけでreadyへ進めてはいけません。
直接アップロードではAPIがファイル本体を受け取らないため、保存後のオブジェクトを基準に完了を確定します。

署名URLを知っている利用者は、有効期間内なら許可された操作を実行できます。
URLをログや解析ツールへ不用意に残さず、対象キー、操作、有効期限を必要最小限に絞ります。

ダウンロードとRangeの扱い

Rangeリクエストは、クライアントがファイルの一部分だけを求めるHTTPリクエストです。
動画のシークや中断したダウンロードの再開では、サーバーがRangeを解釈し、該当範囲を206 Partial Contentで返します。

現在のStarletteのFileResponseはRangeリクエストに対応していますが、実際の挙動は採用バージョンでテストします。
APIから配信する場合も、URLの名前をそのままファイルパスへ結合せず、ファイルIDから認可済みの保存名を引きます。

from uuid import UUID
from fastapi import HTTPException
from fastapi.responses import FileResponse


@router.get("/download/{file_id}")
async def download(file_id: UUID):
    metadata = lookup_ready_file(file_id)  # DBから認可済み情報を取得
    if metadata is None:
        raise HTTPException(404, "not found")

    path = QUARANTINE / metadata.stored_name
    return FileResponse(
        path,
        media_type=metadata.mime,
        filename=metadata.display_name,
    )

大容量ファイルや配信回数の多いファイルは、ストレージまたはCDNの署名付きGETへ移すほうが、APIサーバーの帯域を使わずに済みます。
保存と配信の役割を詳しく比較する場合は、Amazon S3とCloudFrontの使い分けを参照してください。

JSON Lines、SSE、バイナリなど、応答を逐次返す方式そのものを比較したい場合は、FastAPIのストリーミングAPI設計で選択基準を整理しています。

整合性を確認する三つの値

Content-Length、ETag、暗号学的ハッシュは役割が異なります。
どれか一つを記録するだけではなく、何を確認したいのかに合わせて使い分けます。

API経由なら、読み書きと同時にチェックサムを計算できます。
署名URL経由なら、クライアントにチェックサムを送らせ、ストレージ側の検証結果または保存後に読み取った値と照合してから状態を確定します。

公開までを状態で管理する

ファイル本体とデータベースの更新は同時に終わるとは限りません。
そこで、処理段階を状態として保存し、再試行や障害調査で現在地を判断できるようにします。

ウイルススキャン、サムネイル生成、動画変換はワーカーへ渡します。
ジョブは同じ依頼が届いても結果が重複しないように設計し、再試行回数を超えた処理は手動で再投入できるようにします。

セキュリティの確認項目

• アプリ、リバースプロキシ、ストレージの各層でサイズ上限を設定する。
• 申告されたMIMEタイプと、ファイル内容から判定した形式を照合する。
• 利用者のファイル名を保存パスやオブジェクトキーへ直接使わない。
• アップロード直後は隔離し、検査完了前のダウンロードを許可しない。
• 署名URLの対象、操作、有効期限を絞り、URLそのものを秘密情報として扱う。
• アップロード者、ファイルID、オブジェクトキー、サイズ、チェックサム、状態遷移を追跡できるようにする。
• 不要になった一時ファイル、失敗したアップロード、期限切れの未完了データを削除する。

拡張子、MIMEタイプ、ウイルススキャンは、それぞれ検出できる問題が異なります。
単独の検査を「安全の証明」とせず、入力、隔離、検査、公開、配信の各段階に制約を置きます。

運用で見る指標とテスト

平均値だけでは、少数の巨大ファイルや特定利用者の連続失敗を見落とします。
件数、サイズ分布、処理時間、失敗理由、検査待ち時間、署名URLの失効、保存容量を組み合わせて確認します。

• 上限ちょうどのファイルと、上限を1バイト超えるファイルを送る。
• 拡張子、申告MIME、実際の内容が異なるファイルを送る。
• パス区切りや同名を含むファイル名を送る。
• 同じ完了通知とジョブを複数回送る。
• チェックサムが一致しないアップロードを完了させる。
• 正常なRange、範囲外のRange、If-Rangeを使って応答を確認する。
• 期限切れ、権限外、別オブジェクト用の署名URLを使う。

導入の順序

1. 許可するファイル形式、上限、保存期間、公開条件を決めます。
2. UploadFileを使い、隔離保存、チャンク処理、保存名の生成、失敗時削除を実装します。
3. チェックサム、ウイルススキャン、状態遷移を追加します。
4. 大容量ファイルを署名URLによる直接アップロードへ移します。
5. ダウンロードをストレージまたはCDNへ移し、Rangeとキャッシュを検証します。
6. 失敗理由と待ち時間を監視し、上限とタイムアウトを実測値から調整します。

まとめ

FastAPIで大容量ファイルを安定して扱うには、ファイルを読むコードだけでなく、受信から公開までの経路を設計します。
API経由ではUploadFileとチャンク処理でメモリ使用を抑え、GB級や高頻度の転送は署名URLでストレージへ逃がします。

公開前には、実際のサイズ、形式、チェックサム、スキャン結果を確認します。
配信ではRange対応をテストし、状態遷移、失敗時の削除、監視までそろえて初めて運用可能な仕組みになります。