全行をためずにJSON Linesを出力する：設計原理

一言でいうと

遅延生成・行の境界・公開フィールド・出力の上限を、HTTPのストリームにつなげます。

なぜ必要なのか

管理者がすべての注文をダウンロードすると、サーバーのメモリが急増しました。export関数が、JSONを作る前に、すべての行をリストに集めたからです。行単位の出力に変えたものの、本文の中の改行が実際の行の境界を壊し、内部の原価もそのまま出ていきました。ストリーミングは、戻り値の型を1つ変える作業ではなく、遅延評価と表現の契約を一緒に決める作業です。

どう動くのか

各行を検証し、公開フィールドだけを射影したあと、JSONにエンコードして、行の末尾に改行を1つ付けます。ジェネレーターは、呼び出しただけでは入力を消費せず、1つの項目を取り出すと、1行だけ進みます。出力の上限は、boolを除く整数で検証します。最後に、FastAPIのStreamingResponseにつなげて、Content-Typeと、ダウンロードした行を、もう一度パースします。

입력 iterator → 한 행 검증 → 공개 투영 → JSON + 개행 → StreamingResponse

契約を読んで失敗を予測するワークシート

以下は、実装をまるごと暗記するための答案ではなく、ステップごとのコードレビューです。各変更の断片は、意図的に契約を壊しています。変更したあとでも、正常な例は通ることがある点に注意してください。実行する前に、どの入力・例外・状態を観測すれば違いが表に出るかを予想し、実装したあとで、その予想と結果を比べます。

1. 行の契約を検証する

validate_row(row)は、dictで、idがboolを除く正のint、nameが空でないstrのときに、rowを返します。それ以外は、ValueErrorです。追加の内部フィールドは許可します。

判断の根拠: boolと数値を区別し、空の名前をエラーとして扱います。

レビューする誤った変更の断片:

not isinstance(row.get("id"), int)

この断片が入った関数の公開契約と比べてみてください。成功例1つでは区別できないなら、拒否されるべき入力や、失敗したあとの状態を観測の対象に選びます。

2. 公開する行だけを作る

project(row)は、validate_rowのあとで、idとnameだけを持つ新しいdictを返します。元の内部フィールドは、そのまま保存します。

判断の根拠: エクスポートの経路にも、通常のAPIと同じ公開フィールドのポリシーを適用する必要があります。

レビューする誤った変更の断片:

"name":row["name"], "internal_cost":row.get("internal_cost")}

3. 行の境界を保ちながらエンコードする

encode_line(row)は、projectの結果を、ensure_ascii=False、separators=(',',':')、sort_keys=TrueでJSONエンコードして、最後に' 'を1つ付けたstrです。nameの中の改行は、JSONのエスケープでなければなりません。

判断の根拠: 文字列の連結でJSONを作ると、引用符と改行で形式が壊れます。

レビューする誤った変更の断片:

ensure_ascii=True

4. 出力の件数の上限を検証する

validate_max(value)は、boolを除くintの1–1000だけをそのまま返し、それ以外はValueErrorです。

判断の根拠: 無限の入力を、誤って最後まで読まないように、呼び出し側に上限を要求します。

レビューする誤った変更の断片:

<= 1001

5. 必要な行だけを消費する

take_rows(rows, maximum)は、isliceなどで最大maximum個だけを遅延して返すiteratorです。呼び出したときにmaximumを検証し、1回nextすると、入力を1回だけ消費します。

判断の根拠: list(rows)に変換した瞬間、無限の入力や大容量の入力を処理できなくなります。

レビューする誤った変更の断片:

islice(list(rows), validate_max(maximum))

6. 行を遅延してシリアライズする

json_lines(rows, maximum=100)は、take_rowsから受け取った行ごとに、encode_lineをyieldします。すべてをまとめた文字列やリストは返しません。

判断の根拠: オブジェクトの選択と、表現の変換を、それぞれ遅延のステップとして維持します。

レビューする誤った変更の断片:

for row in list(take_rows(rows, maximum)):

7. ダウンロードした行を再び検証する

decode_lines(text)は、splitlinesの空でない各行を、json.loadsしたあとでvalidate_rowして、リストとして返します。空の文字列は[]、空である途中の行は、ValueErrorです。

判断の根拠: 空のファイルと、形式が壊れた空のレコードを区別します。

レビューする誤った変更の断片:

continue

8. HTTPのダウンロードを完成させる

create_app(rows)は、GET /exportで、json_lines(rows, 100)を、application/x-ndjsonのStreamingResponseとして返します。rowsは、再び走査できるリストです。内部フィールドがなく、各行の内容と順序を保存する必要があります。

判断の根拠: Content-Typeだけをストリーミングとして書いて、内部では全体を集めていないかを、ジェネレーターのテストと一緒に確認します。

レビューする誤った変更の断片:

media_type="application/json"

現場での姿

TestClientは、レスポンスをバッファリングするので、ネットワークの最初のバイトの遅延や、メモリの上限全体は、証明しません。遅延評価をしているかどうかは、別のカウント用のジェネレーターで検査します。ストリームが始まったあとで不正な行に出会うと、通常のエラーのJSONでステータスを変えるのが難しくなります。実際のサービスは、事前の検証・行ごとのエラーの形式・中断のポリシーのうち、何を選ぶかを決める必要があります。

次のラボですること

8つのステップが、1つの実行可能な成果物につながります。行の契約を検証する → 公開する行だけを作る → 行の境界を保ちながらエンコードする → 出力の件数の上限を検証する → 必要な行だけを消費する → 行を遅延してシリアライズする → ダウンロードした行を再び検証する → HTTPのダウンロードを完成させる、という流れです。

各ステップは、関数やファイルが存在するという事実ではなく、実際の戻り値・例外・状態の変化を検査します。正解を見たあとには、わざと境界の比較や後始末のコードを変えて、どのテストが失敗するかを確認してください。前のテストが次のステップでも維持される理由を説明し、このラボが保証しない本番の条件を1つ書いてみてください。