FAST APIでPostされたjsonをcsvとして保存する方法

FastAPIによる動的CSVロガーの実装レポート

このドキュメントは、FastAPIを使用してPOSTされたJSONデータをCSVファイルに保存するプログラムの仕様と使い方をまとめたものです。
このプログラムは、POSTされるJSONに新しいキーが含まれていた場合、CSVファイルに動的に列を追加してデータを保存する機能を持ちます。

1. ライブラリのインストール手順

このプログラムを実行するには、FastAPI、Uvicorn（ASGIサーバー）、Pandas（データ操作ライブラリ）が必要です。
以下のコマンドをターミナルで実行し、必要なライブラリをすべてインストールしてください。

pip install "fastapi[all]" pandas

• fastapi[all] は、FastAPI本体に加えてサーバー実行に必要な uvicorn などもまとめてインストールします。
• pandas は、CSVの読み書きとデータフレームの操作に使用します。

2. FastAPIの起動コマンド

プログラムファイル（main.py）があるディレクトリで、以下のコマンドを実行してFastAPIサーバーを起動します。

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

• main: main.py ファイルを指します。
• app: main.py 内で定義された app = FastAPI() のインスタンスを指します。
• --reload: ソースコードが変更されるたびにサーバーを自動で再起動する便利なオプションです。

サーバーが正常に起動すると、http://(IPアドレス):8000 でリクエストを待ち受けます。

3. サンプルリクエストの送付方法

curl コマンドやAPIテストツール（Postmanなど）を使用して、http://(IPアドレス):8000/log エンドポイントにJSONデータをPOSTします。
以下は curl を使用した実行例です。

シナリオ1: 最初のデータPOST

最初のデータをPOSTすると、data フォルダと当日の日付のCSVファイル（例: data/2023-10-27.csv）が新規作成されます。

コマンド:

curl -X POST "http://(IPアドレス):8000/log" \
-H "Content-Type: application/json" \
-d '{"user": "Alice", "action": "login"}'

生成されるCSV (data/YYYY-MM-DD.csv):

posted_datetime,user,action
2023-10-27T14:10:20.123456,Alice,login

シナリオ2: 新しいキーを含むデータをPOST

次に、既存のキーに加えて新しいキー (location) を含むデータをPOSTします。

コマンド:

curl -X POST "http://(IPアドレス):8000/log" \
-H "Content-Type: application/json" \
-d '{"user": "Bob", "action": "view_page", "location": "Tokyo"}'

更新後のCSV (data/YYYY-MM-DD.csv):
CSVファイルが読み込まれ、location 列が新しく追加された後、ファイル全体が上書き保存されます。最初のデータ（Alice）の location は空欄になります。

posted_datetime,user,action,location
2023-10-27T14:10:20.123456,Alice,login,
2023-10-27T14:12:30.654321,Bob,view_page,Tokyo

4. mainプログラム

以下のPythonコードを main.py という名前で保存してください。

import pandas as pd
from datetime import datetime
from pathlib import Path
from typing import Dict, Any

from fastapi import FastAPI

# --- 初期設定 ---

# このプログラムがあるフォルダを基準にする
BASE_DIR = Path(__file__).parent

# dataフォルダのパスを定義し、存在しなければ作成する
DATA_DIR = BASE_DIR / "data"
DATA_DIR.mkdir(parents=True, exist_ok=True)

# FastAPIアプリケーションのインスタンスを作成
app = FastAPI()

# --- FastAPIのエンドポイント定義 ---

@app.post("/log")
async def log_dynamic_columns(data: Dict[str, Any]):
    """
    POSTされたJSONデータをCSVに保存する。
    新しいキーが含まれている場合は、CSVに新しい列を動的に追加する。
    """
    # その日の日付からファイル名を生成 (例: 2023-10-27.csv)
    today = datetime.now().strftime('%Y-%m-%d')
    file_path = DATA_DIR / f"{today}.csv"

    # POSTされた日時とJSONデータを辞書にまとめる
    row_to_write = {
        "posted_datetime": datetime.now().isoformat(),
        **data
    }

    try:
        # --- 動的列追加のロジック ---

        # 1. 既存のCSVファイルを読み込む (存在しない場合は空のDataFrameを作成)
        if file_path.exists() and file_path.stat().st_size > 0:
            # 既存のデータをDataFrameとして読み込む
            # keep_default_na=Falseとna_values=['']で空文字列を正しく扱う
            existing_df = pd.read_csv(file_path, keep_default_na=False, na_values=[''])
        else:
            # ファイルが存在しない場合は、空のDataFrameを準備
            existing_df = pd.DataFrame()

        # 2. POSTされた新しいデータを1行のDataFrameに変換
        new_row_df = pd.DataFrame([row_to_write])

        # 3. 既存のデータと新しいデータを結合する
        # pandas.concatが、新しい列を自動的に検出し、
        # 存在しない値はNaN(Not a Number)として自動で埋めてくれる。
        combined_df = pd.concat([existing_df, new_row_df], ignore_index=True)

        # 4. 結合したDataFrameをCSVファイルに上書き保存する
        # index=False は、DataFrameのインデックス(0, 1, 2...)がCSVに保存されるのを防ぐ
        combined_df.to_csv(file_path, index=False, encoding='utf-8')

        return {
            "message": "Data saved successfully. Columns may have been updated.",
            "file": str(file_path)
        }

    except Exception as e:
        return {"error": str(e)}