📂

FastAPIでファイルアップロード

2024/12/12に公開

API

FastAPI

sqlmodel

tech

FastAPIでファイルアップロードのAPIを作ります。

構築の概要

FastAPIのUploadFileを利用し、リクエストで画像ファイルを受け取り、ローカルディレクトリまたはクラウドストレージに保存する処理を追加
その後、保存したデータのファイルパスをPressReleaseのSchema内のthumbnail_1, thumbnail_2, thumbnail_3などに割り当てます。

Schemaの設定

想定しているSchemaは以下のとおりです。

コードはこちら

from sqlmodel import SQLModel, Field, Relationship, AutoString
from typing import Optional, TYPE_CHECKING
import uuid
from datetime import datetime

if TYPE_CHECKING:
    from .user import User


class BasePressRelease(SQLModel):
    year: str = Field(nullable=False) # 年度
    title: str = Field(nullable = False, unique = False)
    description: Optional[str] = Field(nullable=True, default=None) # 説明、リード文
    content: Optional[str] = Field(nullable=True, default=None) # 内容
    etc: Optional[str] = Field(nullable=True, default=None) # その他
    release_date: Optional[datetime] = Field(default=None)
    author_name: Optional[str] = Field(nullable=True, default=None)
    author_department: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_post_name: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_address_number: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_address: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_tel_num: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_fax_num: Optional[str] = Field(nullable=True, default=None) # 選択式
    author_email: Optional[str] =Field(sa_type=AutoString, nullable=True, default=None)
    thumbnail_1: Optional[str] = Field(nullable=True, default=None)
    thumbnail_2: Optional[str] = Field(nullable=True, default=None)
    thumbnail_3: Optional[str] = Field(nullable=True, default=None)
    file_1: Optional[str] = Field(nullable=True, default=None)
    file_2: Optional[str] = Field(nullable=True, default=None)
    file_3: Optional[str] = Field(nullable=True, default=None)
    is_draft: bool = Field(nullable=True, default=True)

class PressRelease(BasePressRelease, table = True):
    id: Optional[int] = Field(default = None, primary_key = True)
    uuid: str = Field(default_factory = uuid.uuid4, nullable = False)
    created_at: datetime = Field(default = datetime.now(), nullable = False)
    updated_at: datetime = Field(default_factory = datetime.now, nullable = False, sa_column_kwargs = {'onupdate': datetime.now})

    user_id: Optional[int] = Field(default=None, foreign_key="user.id", ondelete="CASCADE")
    user: Optional["User"] = Relationship(back_populates="press_releases")

ファイルアップロードAPI

以下のコードがファイルアップロード用のAPIになります。

コードはこちら

@router.post("/upload_images")
def upload_images(images: list[UploadFile] = File(...)):
    image_paths = []
    try:
        for image in images:
            image_path = f"uploads/thumbnails/{uuid.uuid4()}_{image.filename}"
            with open(image_path, "wb") as file_object:
                shutil.copyfileobj(image.file, file_object)
            image_paths.append(image_path)
        return JSONResponse(content={"image_paths": image_paths})
    except Exception as e:
        print(f"Error during image processing: {e}")
        return JSONResponse(
            content={"error": "画像ファイルのアップロードに失敗しました。{e}"},
            status_code=500
        )

処理の流れ

FastAPIのUploadFileを使用してファイルを受け取ります。

今回はlist形式でファイルを受け取ることから、image_pathsという空の配列を設け、これに追加していきます。

imagesはlist形式のため、for文で値を一つずつ取り出して処理をします。
image_pathには、uuid.uuid4と投稿されたfileのfilenameを組み合わせて保存先を設定しています。

with open(image_path, "wb") as file_object:
ファイルを開いて操作するためのコンテキストを作成します。

【open関数】
第一引数image_pathは保存先のファイルパスです。
第二引数 "wb" はバイナリ書き込みモードを指定します。
【"wb"の意味】
w: 書き込みモード。既存の内容は上書きされます。
b: バイナリモード。画像や動画のような非テキストデータに使用します。
【with文】
ファイル操作を安全に管理するための構文です。
with 文を使うと、処理終了後に自動的にファイルが閉じられます（file_object.close() の呼び出しが不要）。

shutil.copyfileobj(image.file, file_object)
shutilモジュールは、ファイルやディレクトリの操作を提供する標準ライブラリです。
copyfileobjは、ファイルオブジェクト間でデータをコピーするための関数です。

第一引数: コピー元のファイルオブジェクト (image.file)。
第二引数: コピー先のファイルオブジェクト (file_object)。
動作:image.file からバイナリデータを読み取り、file_object に書き込みます。一度に全データをコピーするため、効率的です。

5.appendでimage_pathsにファイルパスを追記し、最終的にURLが追加されたimapge_pathsをreturnします

【参考】

ファイルの消去

上記のコードだけでは、ファイルをアップロードするだけで、その後紐づいているデータ（ここではPress Releaseデータ）が削除されたとしても、ファイルはそのまま残り続けてしまいます。
データ容量だけ増えていく仕組みなので、Press Releaseデータが削除されたときに、紐づいているファイルも一緒に削除する仕組みを構築します。

ファイル消去のAPI

直接的にファイルのみを削除するAPIではありませんが、以下のコードとなります。

コードはこちら

@router.delete("/delete_press_release/{press_uuid}")
def delete_press(
    *,
    session: Session=Depends(get_session),
    press_uuid: str
):
    press_release = get_press_by_uuid(session, press_uuid)

    if not press_release:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Press_release not found."
        )
    return delete_press_release(session, press_uuid)

ファイル消去のロジック

ファイル削除を含めた処理については、以下のコードのとおりです。

コードはこちら

# ファイル削除のロジック
def delete_file_if_exists(delete_press_release: PressRelease):
    delete_file_path = [
        delete_press_release.thumbnail_1,
        delete_press_release.thumbnail_2,
        delete_press_release.thumbnail_3,
        delete_press_release.file_1,
        delete_press_release.file_2,
        delete_press_release.file_3,
    ]

    """ファイルが存在すれば削除する"""
    for file_path in delete_file_path:
        if file_path and os.path.exists(file_path):
            os.remove(file_path)
            
def delete_press_release(session, uuid):
    delete_press_release = get_press_by_uuid(session, uuid)
    
    if not delete_press_release:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail = "Press Release not found."
        )
    
    if delete_press_release:
        # 関連ファイルの削除
        delete_file_if_exists(delete_press_release)
    
    session.delete(delete_press_release)
    session.commit()
    return {"message": "Press Release was deleted."}

これでPress Releaseデータが削除されたときには、アップロードされ、紐づいているデータも合わせて削除されるようになります。

構築の概要

Schemaの設定

ファイルアップロードAPI

処理の流れ

ファイルの消去

ファイル消去のAPI

ファイル消去のロジック

Discussion