FastAPIを使ってメモアプリを作ってみた

TL;DR

前回はFlaskでMPA（Multi Page Application）を書きましたが、今回はFastAPIを用いてSPA（Single Page Application）のバックエンドを書きつつ、将来DDD（Domain Driven Development）をするため、どのようなことを意識したらいいのか考えてみました。

以下のリポジトリにサンプルコードを公開しました。フォルダ構成や記事の中で出てこないファイルなどはこちらをご覧ください。

サンプルコード→ https://github.com/hiseumn/sample-memo-fastapi-backend

また、参考資料に前回の記事を掲載しています。一通り動くところまでの手順はそちらに書いたため、今回は省略します。

必要なライブラリのインストール

今回もパッケージマネージャーはRyeを使用しています。

SQLAlchemy

今回はSQL文を直接書かず、ORMを使用しました。

terminal

rye add sqlalchemy

FastAPI

SPAのフロントエンドとの連携には通常REST APIでの連携となり、パラメーターやレスポンスにはJSON形式が用いられます。このためJSONオブジェクトをPythonのクラスオブジェクトに変換したり、バリデーションをしてくれるPydanticとの連携が前提となっているFastAPIを用います。Flaskでも同様のことができますが、高速なパフォーマンス・豊富な自動生成機能・非同期処理のサポートなど様々なメリットを享受できるため、FastAPIを用います。

terminal

rye add fastapi

uuid

保存するメモのIDとしてUUIDを使用しました。IDを時系列でソートしたかったので、ビルトインのv4ではなくv7を使用するため、以下のライブラリをインストールしています。

terminal

rye add uuid6

主なファイル構成

フォルダの構成は以下のとおりです。

src
    └── sample_fastapi_memo
        ├── app.py
        ├── database.py
        ├── functions.py
        ├── routers.py
        └── schemas.py

app.py

routerをincludeしているだけですが、アプリケーションが対応するドメインが増えたら、ここでそのドメインへのrouterを追記していきます。

python

from routers import router
from fastapi import FastAPI

app = FastAPI()
app.include_router(router)

database.py

DDDで言うインフラ層に相当します。PostgreSQLとのやりとりはここにまとめます。

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Column
from uuid6 import uuid7
from sqlalchemy_utils import UUIDType
from sqlalchemy.types import Text

Base = declarative_base()

engine=create_engine(f"postgresql+psycopg://postgres:postgres@localhost:5432/memo")

def get_db_session() -> scoped_session:
    """ 新しいDBコネクションを返す
    """
    return scoped_session(sessionmaker(bind=engine))

class Memo(Base):
    __tablename__ = "memo"  # テーブル名を指定
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid7)
    memo = Column(Text)

    def memo_result(self):  # メモの内容を返す
        return "{self.id} {self.memo}"
    

※データベースに接続するためのIDとパスワードをハードコーディングすることは、セキュリティリスクになりかねないため真似しないでください。

functions.py

今回はあまり厳密に分けていませんが、DDDで言うユースケースとドメインの部分に相当します。このアプリの実際のロジックをここにまとめます。

python

from sqlalchemy import desc
from sqlalchemy.orm import scoped_session
from database import Memo
from schemas import InputMemo as InputMemoSchema, Memo as MemoSchema
from uuid6 import uuid7

def add_memo(memo: InputMemoSchema, session: scoped_session) -> Memo | None:
    memo = Memo(id=uuid7(), memo=memo.memo)
    session.add(memo)
    session.commit()
    session.refresh(memo)
    return memo


def get_memo(session: scoped_session):
    return session.query(Memo).order_by(desc('id')).limit(20).all()

routers.py

APIのエンドポイントを記述します。ここではビジネスロジックはできる限り書かないようにしています。

python

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import scoped_session
import functions
from database import get_db_session
from schemas import InputMemo, Memo

router = APIRouter()


@router.get("/memo", tags=["/memo"])
def get_memo(session: scoped_session = Depends(get_db_session)) -> list[Memo]:
    memo = functions.get_memo(session)
    return list(map(Memo.model_validate, memo))


@router.post("/memo", tags=["/memo"])
def add_memo(memo: InputMemo, session: scoped_session = Depends(get_db_session)) -> Memo:
    memo = functions.add_memo(memo, session)
    return Memo.model_validate(memo)

schemas.py

似たようなクラスがdatabase.pyにもありますが、こちらはAPIのエンドポイントに関するリクエストとレスポンスのパラメーターを定義しています。InputMemoとMemoに分けたのは、PostのパラメーターとGetのレスポンスでは同じMemoでも必須となる項目が違うからです。

python

from uuid import UUID
from pydantic import BaseModel, ConfigDict


class InputMemo(BaseModel):
    memo: str
    model_config = ConfigDict(from_attributes=True)


class Memo(InputMemo):
    id: UUID

##　参考資料

以上