FastAPIのストリーミングAPI設計：JSON Lines、SSE、バイナリの選び方

FastAPIで処理結果を少しずつ返したいとき、最初に決めるべきなのはコードの書き方ではなく、クライアントが何を受け取り、どのように再利用するかです。

構造化データの列ならJSON Lines、ブラウザへの通知ならServer-Sent Events（SSE）、ファイルや生のバイト列ならStreamingResponseが基本候補になります。

この記事では、FastAPIによるAPI設計の基本を踏まえ、三つの方式を実装と運用の両面から選び分けます。

三つの配信方式を用途で選ぶ

ストリーミングとは、すべてのデータが完成するまで待たず、準備できた項目やチャンクからクライアントへ送る方式です。

FastAPIの公式ドキュメントでは、構造化データ向けのJSON Lines、イベント通知向けのSSE、生の文字列やバイナリ向けのStreamingResponseが別々に説明されています。

FastAPIのストリーミング方式の選択基準

方式	向いている用途	主な形式	設計上の焦点
JSON Lines	検索結果、処理結果、ログなどの構造化された列	application/jsonl	一行ごとのスキーマと完了条件
SSE	進捗、通知、生成処理の途中結果などをブラウザへ送る用途	text/event-stream	イベントID、再接続、再送範囲
StreamingResponse	大きなファイル、音声、動画、生の文字列	用途に合うメディアタイプ	チャンク境界、リソース解放、切断時の処理

三つは同じ「逐次配信」でも、データの境界とクライアント側の扱いが異なります。

サーバーが送れる形式から選ぶのではなく、受信側が項目単位、イベント単位、バイト列のどれとして処理するかで決めると、API契約が曖昧になりません。

JSON Linesは構造化データの列に向く

FastAPIのJSON Lines公式ガイドによると、JSON Linesは一行に一つのJSON値を置き、前の項目をクライアントが処理している間にも次の項目を生成できる形式です。

通常のJSON配列のように全体を角括弧で囲まないため、クライアントは改行を境界として一件ずつ読み進められます。

from collections.abc import AsyncIterable

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Progress(BaseModel):
    step: int
    status: str


@app.get("/jobs/{job_id}/progress")
async def stream_progress(job_id: str) -> AsyncIterable[Progress]:
    for step in range(1, 4):
        yield Progress(step=step, status="running")

戻り値をAsyncIterable[Progress]と宣言すると、FastAPIはPydanticを使って各項目を検証し、OpenAPIへ記録し、シリアライズします。

同期処理しか持たないデータ源では、通常のdefとIterable[Progress]も使えるため、形だけasync defへ変更する必要はありません。

ただし、JSON Linesは「一件ずつ読めるJSON」であり、一つのJSON配列ではありません。

既存クライアントがレスポンス全体を一度にJSONとして解析している場合は、受信処理も行単位へ変更する必要があります。

SSEはブラウザへ意味のあるイベントを送る

FastAPIのSSE公式ガイドでは、EventSourceResponseを指定し、パス操作関数から値をyieldする実装が示されています。

SSEはdataだけでなく、event、id、retryといったフィールドを持てるため、通知の種類や再接続方針をAPI契約へ含められます。

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()


@app.get("/jobs/{job_id}/events", response_class=EventSourceResponse)
async def stream_events(job_id: str) -> AsyncIterable[ServerSentEvent]:
    for sequence in range(1, 4):
        yield ServerSentEvent(
            data={"progress": sequence * 25},
            event="progress",
            id=str(sequence),
            retry=5000,
        )

接続が切れたブラウザは、最後に受信したidをLast-Event-IDヘッダーで送れます。

しかし、IDを付けるだけでは再開できず、サーバー側に「そのID以降を再現できるデータ」が必要です。

保持期間、欠番、期限切れ時の応答を先に決めておくと、再接続後の重複や取りこぼしをクライアントが扱えます。

FastAPIはSSEについて、無通信時の15秒ごとのpingコメント、Cache-Control: no-cache、X-Accel-Buffering: noを既定で扱います。

これらは長時間接続を支える助けになりますが、CDN、ロードバランサー、リバースプロキシを含む本番経路で、タイムアウトとバッファリングの挙動を確認する作業は残ります。

StreamingResponseは生のチャンクをそのまま送る

FastAPIのStream Data公式ガイドは、構造化されたJSONではなく、生の文字列やバイナリを送る用途にStreamingResponseを示しています。

FastAPIは各チャンクをJSONへ変換しないため、ファイル、音声、動画、独自形式のデータを用途に合うメディアタイプで返せます。

from collections.abc import Iterable
from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/exports/report", response_class=StreamingResponse)
def download_report() -> Iterable[bytes]:
    path = Path("report.zip")
    with path.open("rb") as source:
        while chunk := source.read(64 * 1024):
            yield chunk

実ファイルを読む場合は、ジェネレーターが終了したときにファイルを閉じられるよう、withブロックで管理します。

メディアタイプ、ファイル名を示すヘッダー、Rangeリクエストの要否は別の設計事項であり、StreamingResponseを選ぶだけでダウンロード要件がすべて満たされるわけではありません。

ストリーミングを速さの万能薬にしない

逐次配信は、最初の項目が届くまでの待ち時間や、全件をメモリに載せる負担を減らせる場合があります。

一方で、データベース照会、外部API、シリアライズ、ネットワーク帯域が遅ければ、総処理時間そのものは短くなりません。

遅いクライアントを待つ接続が増えるため、同時接続数、タイムアウト、ワーカー数、接続先の上限を合わせて測る必要があります。

HTTPクライアントや接続プールをストリームごとに作らず共有する場合は、FastAPIのlifespanによる共有リソース管理も確認しておくと、初期化と解放の責任を分けられます。

運用前に確認したい設計項目

• データ境界：改行、SSEイベント、固定または可変チャンクのどれで一単位を示すか。
• 終了表現：正常終了、途中エラー、利用者による中断をクライアントがどう区別するか。
• 再接続：どこから再開できるか、同じ項目が再送されても安全か、履歴をいつまで保持するか。
• 認証：長時間接続中に認証期限が切れた場合の扱いと、接続開始時の認可範囲をどう固定するか。
• 観測：接続数、継続時間、最初の項目までの時間、途中切断、完了件数を記録できるか。
• 経路試験：アプリ単体だけでなく、実際のプロキシやCDNを通して逐次到着と切断を確認したか。

テストでは、ステータスコードだけでなく、メディアタイプ、最初の項目、複数項目の境界、完了、途中切断を分けて確認します。

SSEでは、既知のLast-Event-IDから期待したイベントを再送できるかも独立したケースにします。

よくある質問

SSEとJSON Linesは同じものですか

どちらもテキストを逐次送れますが、同じ形式ではありません。

JSON Linesは一行ごとのJSON値を扱い、SSEはdataやidなどを持つイベント形式を扱います。

ストリーミングにはasync defが必須ですか

必須ではありません。

FastAPIの公式ガイドは、非同期イテレーターに加えて、通常のdefから同期イテレーターを返す例も示しています。

通常のJSONレスポンスを置き換えるべきですか

クライアントが全件をそろえてから処理するなら、通常のJSON配列のほうが契約もテストも単純です。

最初の結果を早く使いたい、件数が大きい、継続的に更新を届けたいという要件があるときに、ストリーミングを選びます。

SSEでidを付ければ取りこぼしは防げますか

IDだけでは保証できません。

サーバーがイベント履歴を保持し、Last-Event-IDに対応する位置から再送できる場合に、再開の仕組みとして機能します。

選び方の結論

構造化された列はJSON Lines、ブラウザへ意味のある通知を送るならSSE、生の文字列やバイナリならStreamingResponseを起点にします。

そのうえで、終了、再接続、認証、監視、実経路でのテストまでAPI契約へ落とし込むと、試作で動いたストリームを運用できる機能へ育てられます。

参考資料

• FastAPI公式「Stream JSON Lines」
• FastAPI公式「Server-Sent Events (SSE)」
• FastAPI公式「Stream Data」

この記事に関連する株式会社greedenの取り組み

ストリーミングAPIは、形式選びだけでなく再接続や監視、負荷試験まで含めた設計が欠かせません。株式会社greedenは、要件定義からAPI開発、テスト、保守改善まで一気通貫でWebシステム開発を支援しています。