FastAPIでCSVファイルアップロードして辞書化まで

FastAPIでcsvアップロードしてデータを読み込むなら下記の方法が簡単

概要

1. アップロード
2. 一時ファイル化してパスを取得
3. csvライブラリで辞書化

嵌まった所

• アップロード直後はSpooledTemporaryFileなのでそのままcsvライブラリを利用できなかった（できるかもしれないがわからなかった）対策としてNamedTemporaryFileを使ってパスを取得してwith openした
• upload_file.file.read()で出来なくもないが辞書化が煩雑だった

アップロードするCSVファイル

Id,Name,Age,email
1,John,23,[e-mail]
2,Jose,44,[e-mail]
3,Tonny,29,[e-mail]
4,Mike,35,[e-mail]

実装

import csv
import shutil
import pprint

from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import List

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status

app = FastAPI()

KB = 1024
MB = 1024 * KB

@app.post("/upload/")
def upload_file(upload_file: UploadFile = File(...)):

    # ファイルサイズ検証
    upload_file.file.seek(0, 2) # シークしてサイズ検証
    file_size = upload_file.file.tell()
    if file_size > 20 * MB:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="アップロードファイルは20MB制限です",
        )
    else:
        upload_file.file.seek(0) # シークを戻す

    tmp_path: Path = ""
    try:
        suffix = Path(upload_file.filename).suffix
        with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            shutil.copyfileobj(upload_file.file, tmp)
            tmp_path = Path(tmp.name)
            print(tmp_path)
    except Exception as e:
        print(f"一時ファイル作成: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="一時ファイル作成できません",
        )
    finally:
        upload_file.file.close()

    with open(tmp_path, "r", encoding="utf-8_sig") as f:
        reader = csv.DictReader(f)
        pprint(list(reader))

    return {"message": "アップロード完了"}

結果

[
  {'Id': '1', 'Name': 'John', 'Age': '23', 'email': '[e-mail]'},
  {'Id': '2', 'Name': 'Jose', 'Age': '44', 'email': '[e-mail]'},
  {'Id': '3', 'Name': 'Tonny', 'Age': '29', 'email': '[e-mail]'},
  {'Id': '4', 'Name': 'Mike', 'Age': '35', 'email': '[e-mail]'}
]